跳到主要内容

异步 I/O 编程

信息
2024年8月10日 · ·

I/O 密集型应用程序大大提升系统多任务处理能力;

异步 I/O 模型

一个消息循环,主线程在消息循环中不断重复 读取消息-处理消息

# 获取线程池
loop = get_event_loop()
while True:
# 接收事件消息
event = loop.get_event()
# 处理事件消息
process_event(event)

当遇到 I/O 操作,代码只会发出 I/O 请求,不等待 I/O 结果,当本轮消息结束,下轮接收消息收到 I/O 完成时,处理 I/O 结果;

1. 协程

调度策略由程序员自己编写,在用户态完成创建、切换、销毁,通过协作而非抢占,对内核来说不可见的 用户空间线程

协程的本质是控制流的主动让出(yield)和恢复(resume)机制

  • 子程序,又叫函数,在所有语言都是层级调用,通过栈实现,一个线程就是执行一个子程序,子程序调用总是一个入口,一次返回,调用顺序明确;

  • Coroutine,Python 对协程的支持通过 generator 实现,执行时内部可中断,转而执行别的子程序,再适时返回接着执行(类似 CPU 中断);

协程没有线程切换(抢占式)的开销,且不存在变量冲突,不需要线程锁,效率比多线程高;

1. 生产者-消费者模型(协程版)

def consumer():
r = ''
while True:
# 2. 通过 yield 回传 r 给 send 调用
# 4. 接收 send 的消息 n
n = yield r
if not n:
return
print(f'[CONSUMER] Consuming {n}...')
r = '200 OK'


def produce(c):
# 1. 启动生成器
c.send(None)
n = 0
while n < 5:
n += 1
print(f'[PRODUCER] Producing {n}...')
# 3. 发送消息 n 返回给 yield
# 5. 接收 yield 的结果 r
r = c.send(n)
print(f'[PRODUCER] Consumer return: {r}')
# 6. 关闭生成器
c.close()


# 消费者 - 生成器对象
c = consumer()
produce(c)

2. asyncio

Python 3.4 引入标准库,提供了完善的异步 I/O 支持;

asyncio的编程模型是一个消息循环,首先需要从asyncio获取一个EventLoop的引用,然后把执行的协程扔到EventLoop中执行,从而实现异步 I/O;

import asyncio


# @aysncio.coroutine 把 generator 标记成 coroutine
@asyncio.coroutine
def wget(host):
print('wget %s...' % host)
connect = asyncio.open_connection(host, 80)
# yield from 调用 connect 生成器,并接受 connect 的调用结果
# 主线程并未等待 connect 调用,而是执行 EventLoop 中其他 coroutine
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']
]
# 把 coroutine 扔到 EventLoop 中执行
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

异步操作在coroutine中通过yield from完成;

3. async/await

Python 3.5 引入的针对 coroutine 的新语法;

  • async,替换 @asyncio.coroutine
  • await,替换 yield from

4. aiohttp

  • asyncio,实现了TCPUDPSSL等协议;
  • aiohttp,基于asyncio实现了HTTP框架;

1. 安装

$ pip install aiohttp

2. 示例

import asyncio
from aiohttp import web


async def index(request):
await asyncio.sleep(1)
return web.Response(body=b'<h1>Index</h1>')


async def hello(request):
await asyncio.sleep(1)
text = '<h1>hello, %s!</h1>' % request.match_info['name']
return web.Response(body=text.encode('utf-8'))


async def init(loop):
app = web.Application(loop=loop)
app.router.add_route('GET', '/', index)
app.router.add_route('GET', '/hello/{name}', hello)
# 利用 asyncio 创建 TCP 服务
srv = await loop.create_server(app.make_handler(), '', 8000)
print('server started at http://localhost:8000...')
return srv


loop = asyncio.get_event_loop()
loop.run_until_complete(init(loop))
loop.run_forever()

PS:欢迎各路道友阅读评论,感谢道友点赞关注收藏