1.协程
#协程 又称微线程 是一种用户的轻量级线程 程序级别代码控制 就不用加机器 #不同函数 = 不同任务 A函数切到B函数没有进行cpu级别的切换，而是程序级别的切换就是协程 yelied #单线程下多个任务流用协程，比如打电话可以切换,nginx #爽妹给你打电话的时候，她不说话，刘征电话过来时候你可以切过去，这时候要是爽妹说话，就会bibi响 ''' 协程的好处： 无需线程上下文切换的开销 无需原子操作锁定及同步的开销 "原子操作(atomic operation)是不需要synchronized"，所谓原子操作是指不会被线程调度机制打断的操作；这种操作一旦开始，就一直运行到结束，中间不会有任何 context switch （切换到另一个线程）。原子操作可以是一个步骤，也可以是多个操作步骤，但是其顺序是不可以被打乱，或者切割掉只执行部分。视作整体是原子性的核心。 方便切换控制流，简化编程模型 高并发 + 高扩展性 + 低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。 缺点： 无法利用多核资源：协程的本质是个单线程, 它不能同时将 单个CPU 的多个核用上, 协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。 进行阻塞（Blocking）操作（如IO时）会阻塞掉整个程序
2.Greenlet and Gevent
#greenlet 模块
#greenlet是一个用C实现的协程模块，相比与python自带的yield，它可以使你在任意函数之间随意切换，而不需把这个函数先声明为generator
from greenlet import greenlet
from greenlet import greenlet
def test1():
print(12)
#time.sleep(1) #但是遇到IO就会阻塞了，这里延迟了一秒，如果自动切换的话，应该立马执行gr2
gr2.switch()
print(34)
gr2.switch()
def test2():
print(56)
gr1.switch()
print(78)
gr1 = greenlet(test1) #生成协程
gr2 = greenlet(test2) #生成协程
gr1.switch() #启动协程
#但是遇到IO会不会自动切换呢？上面是手动切换的 引出 Gevent
#Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet,
#（接着上面一条）它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。
import gevent
def func1():
print('\033[31;1m李闯在跟海涛搞...\033[0m') #1 1
gevent.sleep(2) #相当于io time.sleep 卡住，看看会不会自动切换，还是等待？
print('\033[31;1m李闯又回去跟继续跟海涛搞...\033[0m') #4 6
def func2():
print('\033[32;1m李闯切换到了跟海龙搞...\033[0m') #2 2
gevent.sleep(1) #这里自动切换的时候上面还在阻塞呢，所以又回来了，所以按 1234标识了走向
print('\033[32;1m李闯搞完了海涛，回来继续跟海龙搞...\033[0m') #3 4
def func3():
') #0 3
gevent.sleep(1)
') #0 5
gevent.joinall([ #joinall等待所有协程结束 这是一个列表
gevent.spawn(func1), #产生协程
gevent.spawn(func2),
gevent.spawn(func3),
])
3.同步与异步的性能区别
#同步与异步的性能区别
import gevent
def task(pid):
"""
Some non-deterministic task
"""
gevent.sleep(0.5)
print('Task %s done' % pid)
def synchronous(): #同步就是串行的效果
for i in range(1, 10):
task(i)
def asynchronous(): #异步就是并发的效果
threads = [gevent.spawn(task, i) for i in range(10)]
gevent.joinall(threads)
print('Synchronous:') #同步
synchronous()
print('Asynchronous:') #异步
asynchronous()
4.爬网页
#url爬网页
import gevent
from urllib.request import urlopen #现在还是阻塞的模式，因为urllib遇到Io不知道这是Io操作，所以需要导入一个gevevt插件，
#相当于打个补丁，就会把urllib 里面涉及IO操作的都改成异步的模式，不阻塞的模式
from gevent import monkey #补丁
monkey.patch_all() #补丁 注意顺序
import time
def pa_web_page(url):
print('get url',url)
req = urlopen(url) #抓取url
data = req.read() #读取结果
print(data)
print('%d bytes received from %s.' % (len(data), url))
t1_start = time.time() #开始时间
pa_web_page("https://www.baidu.com")
pa_web_page("http://www.xiaohuar.com")
print('time close t1',time.time()-t1_start) #做减法
#下面是协程gevent写法，遇到阻塞就会自动切换，节省了时间
t2_start = time.time() #开始时间
gevent.joinall([
gevent.spawn(pa_web_page, 'https://www.baidu.com'), #pa_web_page,函数名 https://www.baidu.com url
gevent.spawn(pa_web_page, 'http://www.xiaohuar.com'),
])
print('time close t2',time.time()-t2_start) #做减法
5.通过gevent实现单线程下的多socket并发
server code
import sys
import socket
import time
import gevent
from gevent import socket, monkey
monkey.patch_all()
def server(port):
s = socket.socket()
s.bind(('0.0.0.0', port))
s.listen(500)
while True:
cli, addr = s.accept()
gevent.spawn(handle_request, cli) #之前写线程sockserver的时候是起一个线程，这里是起协程
# handle_request自己写的方法 所有请求到这个函数区处理
def handle_request(conn):
try:
while True:
data = conn.recv(1024)
print("recv:", data)
conn.send(data)
if not data:
conn.shutdown(socket.SHUT_WR) #相当于断开连接，清空了
except Exception as ex:
print(ex)
finally:
conn.close()
if __name__ == '__main__':
server(8001)
client code
#并发100个链接 如果连接报错，就说明开不起线程了，确实支持大并发了
import socket
import threading
def sock_conn():
client = socket.socket()
client.connect(("localhost",8001))
count = 0
while True:
#msg = input(">>:").strip()
#if len(msg) == 0:continue
client.send( ("hello %s" %count).encode("utf-8"))
data = client.recv(1024)
print("[%s]recv from server:" % threading.get_ident(),data.decode()) #结果
count +=1
client.close()
for i in range(100):
t = threading.Thread(target=sock_conn)
t.start()
6.Select\Poll\Epoll IO多路复用
select
import socket
import select
import queue
server = socket.socket()
server.bind(("localhost",8001))
server.listen(5)
server.setblocking(0) #设置为非堵塞
inputs = [server] #select 维护的列表，也是是传过来的链接 首先是监听自己
msg_queues = {} #字典，为了收取数据作用，理论上应该有2个，一个是收，一个是取
outputs = []
while True:
r_list,w_list,exception_list = select.select(inputs,outputs,inputs) #inputs检测所有socket有没有消息古来 outputs不知道 inputs检测哪些socket有没有错（错误）
#针对 inputs 来返回哪些就绪的列表，所以r_list里面的就已经是就绪的 相当于链接
# print("r_list",r_list)
# print("w_list",w_list)
# print("e_list",exception_list)
for s in r_list: #数据流
if s is server: #这是一个新链接
conn,addr = s.accept() #接收请求 同时可以监听多个请求了
print("got a new conn",conn,addr)
inputs.append(conn) #让select去监测客户端是否有数据过来
msg_queues[conn] = queue.Queue() #为了给客户端返回数据，先创建的数据字典
else:
try:
data = s.recv(1024)
print("recv data from [%s]:[%s]" % (s.getpeername(),data.decode()))
msg_queues[s].put(data)
if s not in outputs:
outputs.append(s) #等下次select的时候，确保w_list的数据能返回给客户端
except ConnectionResetError as e:
print("conn closed.",s.getpeername(),e)
inputs.remove(s) #链接出问题，或意外终止
if s in outputs:
outputs.remove(s)
del msg_queues[s]
for s in w_list: #给客户端返回追备好的数据
try:
data = msg_queues[s].get_nowait()
s.send(data.upper())
except queue.Empty as e:
outputs.remove(s)
selectors select的升级版，自动适应版本执行epool效率更高
#selectors模块 这是自适应的，你系统默认支持的话，就会epool 》pool 》select 相比select更方便 写这个代码默认epool
import selectors
import socket
def accept(sock, mask):
conn, addr = sock.accept() # Should be ready
print('accepted', conn, 'from', addr)
conn.setblocking(False)
sel.register(conn, selectors.EVENT_READ, read)
#监听数据流，如果消息事件来了，调用read方法 注册conn用于监控流了
def read(conn, mask):
data = conn.recv(1000) # Should be ready
if data:
print('echoing', repr(data), 'to', conn)
conn.send(data) # Hope it won't block
else:
print('closing', conn)
sel.unregister(conn) #删除链接清空 跟select remove一样
conn.close()
sock = socket.socket()
sock.bind(('localhost', 8001)) #端口是0-65535 1024系统保留
sock.listen(100)
sock.setblocking(False)
sel = selectors.DefaultSelector()
sel.register(sock, selectors.EVENT_READ, accept) #sock相当于注册，注册一个什么呢，注册一个EVENT_READ读事件 这只是注册呢没有实际监听
#相当于 select.select(inputs,outputs.... EVENT_READ 监听，如果有请求就会调用accept)
while True:
events = sel.select() #如果没有事件就会卡这里，select监听，
for key, mask in events:
callback = key.data #相当于accept内存对象
print(key,mask)
callback(key.fileobj, mask)
7.RabbitMq 进程队列
server
# !/usr/bin/env python
import pika
# credentials = pika.PlainCredentials('alex','alex3714') 假如需要验证的时候用这2条就可以连接
# connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.10.10.140',credentials=credentials))
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.10.10.140'))
channel = connection.channel()
# 声明queue
channel.queue_declare(queue='hello1',durable=True)#durable=True queue队列持久化，rabbitmq重启不会丢失，但是消息会丢
#如果之前这里生命过durable，在recv端也要这样声明
# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange='',
routing_key='hello1',
body='Hello World!1',
properties=pika.BasicProperties( #消息持久化 rabbit重启消息不会丢
delivery_mode=2, # make message persistent #消息持久化abbit重启消息不会丢
))
print(" [x] Sent 'Hello World1!'")
connection.close()
client
# _*_coding:utf-8_*_
__author__ = 'Alex Li'
import pika
import time
# credentials = pika.PlainCredentials('alex','alex3714') 假如需要验证的时候用这2条就可以连接
# connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.10.10.140',credentials=credentials))
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.10.10.140'))
channel = connection.channel()
# You may ask why we declare the queue again ‒ we have already declared it in our previous code.
# We could avoid that if we were sure that the queue already exists. For example if send.py program
# was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
channel.queue_declare(queue='hello1',durable=True) #如果确定这个queue声明过了，可以不用写，但是写上最好，因为不确定send端还是recv先启动
def callback(ch, method, properties, body): #body消息
#ch channel对象 method 声明的一推参数，消息里面的一些属性信息 properties跟随消息传一些参数会在这个里面
print(" [x] Received %r" % body)
# time.sleep(10) #用于测试work queue
channel.basic_qos(prefetch_count=1) #消息公平化，如果有一个消息没有处理完就别给我发新的
channel.basic_consume(callback, #在hello queue里面收取消息执行callback函数
queue='hello1',
#no_ack=True
) #true 默认开启 work queue 这样可以确保即使消息发送的时候中断，也会受到信息no_ack=True这个是关闭了
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #有消息就收，没有消息就会卡住 监听