关于协程
协程 coroutine 微线程,一种用户态的轻量级线程
好处：
无需线程上下文切换的开销
无需原子操作锁定及同步的开销
方便切换控制流，简化编程模型
高并发+高扩展+低成本，一个cup支持上万的协程都不是问题
缺点：
无法利用多核资源，协程的本质是单线程，
进程阻塞blocking操作如io时会阻塞整个程序
单线程下实现并发效果：遇到io就切换
服务器处理模型：
1.一个进程处理一个请求
2.一个线程处理一个请求
3.主进程处理事件队列的请求
事件驱动模型：
多个事件 -> 消息队列 -> 处理线程
事件驱动编程是一种编程范式，这里程序的执行流由外部事件来决定。
它的特点是包含一个事件循环，当外部事件发生时使用回调机制来触发相应的处理。
另外两种常见的编程范式是（单线程）同步以及多线程编程。
异步asynchronous
用户空间application 和 内核空间kernel
进程控制块（Processing Control Block）
参考：
http://www.cnblogs.com/alex3714/articles/5876749.html
io操作两个阶段：内核缓冲区数据准备，拷贝到应用程序地址空间
阻塞io blocking 一直等待
非阻塞io nonblocking 准备数据立刻返回
多路复用 select，poll，epoll
异步io 没有被阻塞
twisted框架
手动切换协程
from greenlet import greenlet def foo1(): print("foo11") gr2.switch() print("foo12") gr2.switch() def foo2(): print("foo21") gr1.switch() print("foo22") gr1 = greenlet(foo1) # 创建协程 gr2 = greenlet(foo2) print("main") gr1.switch() # 切换协程 print("done") """ main foo11 foo21 foo12 foo22 done """
自动切换协程
import gevent # 第三方库 def foo1(): print("foo11") gevent.sleep(2) print("foo12") def foo2(): print("foo21") gevent.sleep(2) print("foo22") def foo3(): print("foo31") gevent.sleep(0) print("foo32") gevent.joinall([ gevent.spawn(foo1), gevent.spawn(foo2), gevent.spawn(foo3) ]) """ foo11 foo21 foo31 foo32 foo12 foo22 """
gevent 多并发socket
import socket import gevent from gevent import monkey monkey.patch_all() # 猴子补丁 def server(port): s = socket.socket() s.bind(("0.0.0.0", port)) s.listen(500) print("服务已启动") while True: conn, addr = s.accept() print(addr) gevent.spawn(handle_request, conn) def handle_request(conn): # 处理请求的协程 try: while True: data = conn.recv(1024) print(conn, data.decode()) conn.send(data) if not data: conn.shutdown(socket.SHUT_WR) except Exception as e: print(e) finally: print("关闭", conn) conn.close() if __name__ == "__main__": server(6969)
select 多并发socket
# fd 文件描述符 import select import socket import queue server = socket.socket() server.setblocking(False) # 不阻塞 server.bind(("localhost", 6969)) server.listen(5) print("服务已开启") inputs = [server, ] # server本身也是一个文件描述符 outputs = [] message_dict = {} while True: # 如果没有任何 fd 就绪，就会阻塞在这里 readable, writeable, exceptional = select.select(inputs, outputs, inputs) print("readable", readable) print("writeable", writeable) print("exceptional", exceptional) for r in readable: if r is server: # server就绪，新连接来了 conn, addr = r.accept() print("已连接", conn) conn.setblocking(False) inputs.append(conn) # 为了不阻塞程序，收到连接对象后放入列表，如果接收到信息，fd就绪 message_dict[conn] = queue.Queue() # 新建消息队列，不立刻返回，逐个处理 else: # 如果不是服务器，就是客户端 data = r.recv(1024) if not data: print("连接断开") else: print("收到数据", data.decode("utf-8")) message_dict[r].put(data) # 放入消息队列 outputs.append(r) # 不影响其他客户端连接，后续处理发送信息 for s in outputs: # 要返回给客户端的连接表 data = message_dict[s].get() s.send(data) outputs.remove(s) # 处理完就删除 for e in exceptional: # 删除异常连接 if e in outputs: outputs.remove(e) inputs.remove(e) del message_dict[e]
selectors 多并发socket
import selectors import socket def accept(server, mask): # 接受连接 conn, addr = server.accept() print("conn:", conn, "addr:", addr, "mask:", mask) conn.setblocking(False) selector.register(conn, selectors.EVENT_READ, action) def action(conn, mask): # 接收数据 data = conn.recv(1024) if data: print("conn:", conn, "data:", data.decode("utf-8")) conn.send(data) else: print("断开连接") conn.close() selector.unregister(conn) server = socket.socket() server.setblocking(False) address = ("localhost", 6969) server.bind(address) server.listen(1000) selector = selectors.DefaultSelector() selector.register(server, selectors.EVENT_READ, accept) # accept回调函数 print("服务启动") while True: events = selector.select() # 默认阻塞，有活动则返回活动列表 for key, mask in events: callback = key.data # accept callback(key.fileobj, mask) # fileobj 文件句柄
异步爬虫(并行)
from urllib import request from gevent import monkey import time, gevent monkey.patch_all() # 把当前程序所有io操作都做上标记 def get_html(url): response = request.urlopen(url) html = response.read() print("receive:", len(html)) urls = [ "https://www.python.org/", "https://www.yahoo.com/", "https://github.com/" ] start_time = time.time() for url in urls: html = get_html(url) end_time = time.time() print("串行：", end_time - start_time) async_start_time = time.time() gevent.joinall([ gevent.spawn(get_html, urls[0]), gevent.spawn(get_html, urls[1]), gevent.spawn(get_html, urls[2]) ]) async_end_time = time.time() print("并行：", async_end_time - async_start_time) """ receive: 48893 receive: 511654 receive: 52225 串行： 4.339118003845215 receive: 48893 receive: 502566 receive: 52223 并行： 1.4489901065826416 """
