Python并发编程之IO模型
五种IO模型
为了更好地了解IO模型,我们需要事先回顾下:同步、异步、阻塞、非阻塞
- 同步(synchronous) IO
- 异步(asynchronous) IO
- 阻塞(blocking) IO
- 非阻塞(non-blocking)IO
五种I/O模型包括:阻塞I/O、非阻塞I/O、信号驱动I/O(不常用)、I/O多路转接、异步I/O。其中,前四个被称为同步I/O。
上五个模型的阻塞程度由低到高为:阻塞I/O > 非阻塞I/O > 多路转接I/O > 信号驱动I/O > 异步I/O,因此他们的效率是由低到高的。
1、阻塞I/O模型
在linux中,默认情况下所有的socket都是blocking,除非特别指定,几乎所有的I/O接口 ( 包括socket接口 ) 都是阻塞型的。
如果所面临的可能同时出现的上千甚至上万次的客户端请求,“线程池”或“连接池”或许可以缓解部分压力,但是不能解决所有问题。总之,多线程模型可以方便高效的解决小规模的服务请求,但面对大规模的服务请求,多线程模型也会遇到瓶颈,可以用非阻塞接口来尝试解决这个问题。
2、非阻塞I/O模型
在非阻塞式I/O中,用户进程其实是需要不断的主动询问kernel数据准备好了没有。但是非阻塞I/O模型绝不被推荐。
非阻塞,不等待。比如创建socket对某个地址进行connect、获取接收数据recv时默认都会等待(连接成功或接收到数据),才执行后续操作。
如果设置setblocking(False),以上两个过程就不再等待,但是会报BlockingIOError的错误,只要捕获即可。
异步,通知,执行完成之后自动执行回调函数或自动执行某些操作(通知)。比如做爬虫中向某个地址baidu。com发送请求,当请求执行完成之后自执行回调函数。
3、多路复用I/O模型(事件驱动)
基于事件循环的异步非阻塞框架:如Twisted框架,scrapy框架(单线程完成并发)。
检测多个socket是否已经发生变化(是否已经连接成功/是否已经获取数据)(可读/可写)IO多路复用作用?
操作系统检测socket是否发生变化,有三种模式:
- select:最多1024个socket;循环去检测。
- poll:不限制监听socket个数;循环去检测(水平触发)。
- epoll:不限制监听socket个数;回调方式(边缘触发)。
Python模块:
- select.select
- select.epoll
基于IO多路复用+socket非阻塞,实现并发请求(一个线程100个请求)
import socket # 创建socket client = socket.socket() # 将原来阻塞的位置变成非阻塞(报错) client.setblocking(False) # 百度创建连接: 阻塞 try: # 执行了但报错了 client.connect(('www.baidu.com',80)) except BlockingIOError as e: pass # 检测到已经连接成功 # 问百度我要什么? client.sendall(b'GET /s?wd=alex HTTP/1.0\r\nhost:www.baidu.com\r\n\r\n') # 我等着接收百度给我的回复 chunk_list = [] while True: # 将原来阻塞的位置变成非阻塞(报错) chunk = client.recv(8096) if not chunk: break chunk_list.append(chunk) body = b''.join(chunk_list) print(body.decode('utf-8'))
selectors模块
#服务端 from socket import * import selectors sel=selectors.DefaultSelector() def accept(server_fileobj,mask): conn,addr=server_fileobj.accept() sel.register(conn,selectors.EVENT_READ,read) def read(conn,mask): try: data=conn.recv(1024) if not data: print('closing',conn) sel.unregister(conn) conn.close() return conn.send(data.upper()+b'_SB') except Exception: print('closing', conn) sel.unregister(conn) conn.close() server_fileobj=socket(AF_INET,SOCK_STREAM) server_fileobj.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) server_fileobj.bind(('127.0.0.1',8088)) server_fileobj.listen(5) server_fileobj.setblocking(False) #设置socket的接口为非阻塞 sel.register(server_fileobj,selectors.EVENT_READ,accept) #相当于网select的读列表里append了一个文件句柄 #server_fileobj,并且绑定了一个回调函数accept while True: events=sel.select() #检测所有的fileobj,是否有完成wait data的 for sel_obj,mask in events: callback=sel_obj.data #callback=accpet callback(sel_obj.fileobj,mask) #accpet(server_fileobj,1) #客户端 from socket import * c=socket(AF_INET,SOCK_STREAM) c.connect(('127.0.0.1',8088)) while True: msg=input('>>: ') if not msg:continue c.send(msg.encode('utf-8')) data=c.recv(1024) print(data.decode('utf-8'))
4、异步I/O
asyncio
是Python 3.4版本引入的标准库,直接内置了对异步IO的支持。
asyncio
的编程模型就是一个消息循环。我们从asyncio
模块中直接获取一个EventLoop
的引用,然后把需要执行的协程扔到EventLoop
中执行,就实现了异步IO。
用asyncio
实现Hello world
代码如下:
import asyncio @asyncio.coroutine def hello(): print("Hello world!") # 异步调用asyncio.sleep(1): r = yield from asyncio.sleep(1) print("Hello again!") # 获取EventLoop: loop = asyncio.get_event_loop() # 执行coroutine loop.run_until_complete(hello()) loop.close()
@asyncio.coroutine
把一个generator标记为coroutine类型,然后,我们就把这个coroutine
扔到EventLoop
中执行。
hello()
会首先打印出Hello world!
,然后,yield from
语法可以让我们方便地调用另一个generator
。由于asyncio.sleep()
也是一个coroutine
,所以线程不会等待asyncio.sleep()
,而是直接中断并执行下一个消息循环。当asyncio.sleep()
返回时,线程就可以从yield from
拿到返回值(此处是None
),然后接着执行下一行语句。
把asyncio.sleep(1)
看成是一个耗时1秒的IO操作,在此期间,主线程并未等待,而是去执行EventLoop
中其他可以执行的coroutine
了,因此可以实现并发执行。
我们用Task封装两个coroutine
试试:
import threading import asyncio @asyncio.coroutine def hello(): print('Hello world! (%s)' % threading.currentThread()) yield from asyncio.sleep(1) print('Hello again! (%s)' % threading.currentThread()) loop = asyncio.get_event_loop() tasks = [hello(), hello()] loop.run_until_complete(asyncio.wait(tasks)) loop.close()
观察执行过程:
Hello world! (<_MainThread(MainThread, started 140735195337472)>) Hello world! (<_MainThread(MainThread, started 140735195337472)>) (暂停约1秒) Hello again! (<_MainThread(MainThread, started 140735195337472)>) Hello again! (<_MainThread(MainThread, started 140735195337472)>)
由打印的当前线程名称可以看出,两个coroutine
是由同一个线程并发执行的。
如果把asyncio.sleep()
换成真正的IO操作,则多个coroutine
就可以由一个线程并发执行。
我们用asyncio
的异步网络连接来获取sina、sohu和163的网站首页:
import asyncio @asyncio.coroutine def wget(host): print('wget %s...' % host) connect = asyncio.open_connection(host, 80) reader, writer = yield from connect header = 'GET / HTTP/1.0\r\nHost: %s\r\n\r\n' % host writer.write(header.encode('utf-8')) yield from writer.drain() while True: line = yield from reader.readline() if line == b'\r\n': break print('%s header > %s' % (host, line.decode('utf-8').rstrip())) # Ignore the body, close the socket writer.close() loop = asyncio.get_event_loop() tasks = [wget(host) for host in ['www.sina.com.cn', 'www.sohu.com', 'www.163.com']] loop.run_until_complete(asyncio.wait(tasks)) loop.close()
执行结果如下:
wget www.sohu.com... wget www.sina.com.cn... wget www.163.com... (等待一段时间) (打印出sohu的header) www.sohu.com header > HTTP/1.1 200 OK www.sohu.com header > Content-Type: text/html ... (打印出sina的header) www.sina.com.cn header > HTTP/1.1 200 OK www.sina.com.cn header > Date: Wed, 20 May 2015 04:56:33 GMT ... (打印出163的header) www.163.com header > HTTP/1.0 302 Moved Temporarily www.163.com header > Server: Cdn Cache Server V2.0 ...
可见3个连接由一个线程通过coroutine
并发完成。
async/await
用asyncio
提供的@asyncio.coroutine
可以把一个generator标记为coroutine类型,然后在coroutine内部用yield from
调用另一个coroutine实现异步操作。
为了简化并更好地标识异步IO,从Python 3.5开始引入了新的语法async
和await
,可以让coroutine的代码更简洁易读。
请注意,async
和await
是针对coroutine的新语法,要使用新的语法,只需要做两步简单的替换:
- 把
@asyncio.coroutine
替换为async
; - 把
yield from
替换为await
。
让我们对比一下上一节的代码:
@asyncio.coroutine def hello(): print("Hello world!") r = yield from asyncio.sleep(1) print("Hello again!")
用新语法重新编写如下:
async def hello(): print("Hello world!") r = await asyncio.sleep(1) print("Hello again!")
剩下的代码保持不变。
小结
asyncio
提供了完善的异步IO支持;
异步操作需要在coroutine
中通过yield from
完成;
多个coroutine
可以封装成一组Task然后并发执行。
到此这篇关于Python并发编程之IO模型的文章就介绍到这了。希望对大家的学习有所帮助,也希望大家多多支持我们。