Python 协程异步IO asyncio 模块介绍

asyncio模块在Python3.4标准库中引入，实现使用协程构建异步并发功能

介绍

• 相关扩展：进程、线程、协程的作用和区别
• asyncio的编程模型核心 事件循环(Event Loop)，本质是一个无限的死循环
  • 程序把函数注册到 事件循环(Event Loop) 上
  • 当满足事件的条件时，调用相应的协程函数，实现异步 IO
• 关键字
  • 函数名前加 async 关键字
  • await 表示在这个地方等待子函数执行完成，再往下执行
    • 在并发操作中，把程序控制权教给主程序，让他分配其他协程执行
  • await 只能在带有 async 关键字的函数中运行

使用场景

一般用在 IO 密集型任务

• 网络请求，比如爬虫、aiohttp
• 文件读取 aiofile
• web 框架 aiohttp、fastapi
• 数据库查询 asyncpg 等

Future

asyncio.Futures 代表还未完成的结果（有可能是一个 Exception），代表还没有做完的事情

import asyncio
future = asyncio.Future()

常见方法：

• cancel() 取消 future 的执行，调度回调函数
• result() 返回 future 代表的结果
• exception() 返回 future 中的 Exception
• add_done_callback(fn) 添加一个回调函数，当 future 执行的时候会调用这个回调函数
• remove_done_callback(fn) 从 call whten done 列表中移除所有 callback 的实例
• set_result(result) 将 future 标为执行完成，并且设置 result 的值
• set_exception(exception) 将 future 标为执行完成，并设置 Exception

Task

class asyncio.Task(coro, *, loop=None, name=None, context=None, eager_start=False)

Task 对象被用来在 事件循环 中运行协程。如果一个协程在等待一个 Future 对象，Task 对象会挂起该协程的执行并等待该 Future 对象完成。当该 Future 对象完成，被打包的协程将恢复执行

示例

先看官方的示例

Hello World

import asyncio

@asyncio.coroutine
def hello_world():
    print("Hello World!")

loop = asyncio.get_event_loop()
# Blocking call which returns when the hello_world() coroutine is done
loop.run_until_complete(hello_world())
loop.close()

说明：

• @asyncio.coroutine 把方法标记为 coroutine 类型，通过 loop.run_until_complete 将其添加到 Event Loop 中执行

Coroutine displaying the current date

import asyncio
import datetime

@asyncio.coroutine
def display_date(loop):
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        yield from asyncio.sleep(1)

loop = asyncio.get_event_loop()
# Blocking call which returns when the display_date() coroutine is done
loop.run_until_complete(display_date(loop))
loop.close()

说明：

• asyncio.sleep(1) 业是一个 coroutine，执行时线程不会等待 asyncio.sleep()，而是直接中断并执行下一个 Event Loop，实现类似于并发执行的效果

Parallel execution of tasks

import asyncio

@asyncio.coroutine
def factorial(name, number):
    f = 1
    for i in range(2, number+1):
        print("Task %s: Compute factorial(%s)..." % (name, i))
        yield from asyncio.sleep(1)
        f *= i
    print("Task %s: factorial(%s) = %s" % (name, number, f))

loop = asyncio.get_event_loop()
tasks = [
    asyncio.ensure_future(factorial("A", 2)),
    asyncio.ensure_future(factorial("B", 3)),
    asyncio.ensure_future(factorial("C", 4))]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

说明：

• 通过 asyncio.wait 实现一组 task 的执行

Future with run_until_complete

import asyncio

@asyncio.coroutine
def slow_operation(future):
    yield from asyncio.sleep(1)
    future.set_result('Future is done!')

def got_result(future):
    print(future.result())
    loop.stop()

loop = asyncio.get_event_loop()
future = asyncio.Future()
asyncio.ensure_future(slow_operation(future))
future.add_done_callback(got_result)
try:
    loop.run_forever()
finally:
    loop.close()

同时调度多个 async 函数

import asyncio
import threading


async def hello(name):
    print("hi %s! (%s)" % (name, threading.get_ident()))
    await asyncio.sleep(1)
    print("hi %s again! (%s)" % (name, threading.get_ident()))
    return name


async def main():
    L = await asyncio.gather(hello("Bob"), hello("Alice"))
    print(L)


asyncio.run(main())