python协程与任务

January 13, 2023

https://docs.python.org/zh-cn/3/library/asyncio-task.html

事件循环

https://docs.python.org/zh-cn/3/library/asyncio-eventloop.html

要运行 Coroutine，需要获取事件循环

import asyncio

loop = asyncio.new_event_loop()  # 创建一个事件循环
asyncio.set_event_loop(loop)  # 将当前线程的事件循环设置为该事件循环
print(loop == asyncio.get_event_loop())  # 当前事件循环就是刚才设置的事件循环
# True

可等待对象

import asyncio
from asyncio import Future, Task
from typing import Coroutine, Awaitable

f"{Coroutine, Future, Task} 都是 {Awaitable} 对象"

async def async_func(x=0.1, /):
    await asyncio.sleep(x)
    return x

coroutine = async_func(1)   # async 函数返回 Coroutine 对象
asyncio.run(coroutine)  # 运行 Coroutine 的最简单方法是使用 asyncio.run() 函数（无需事件循环）

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

loop.run_until_complete(async_func())  # 将 Coroutine 注册到事件循环，并启动事件循环
# 相当于:（asyncio.create_task() 方法接受一个 Coroutine 或 Generator）
# 当一个 Coroutine 通过 asyncio.create_task() 等函数被封装为一个 Task 时，该 Coroutine 会被自动调度执行
task = loop.create_task(async_func())  # 创建 Task。此时的 Task 尚未加入事件循环，状态为 pending
loop.run_until_complete(task)  # 将 Task 注册到事件循环，并启动事件循环。Task 执行完后，状态为 finished


# Task 对象允许设置完成时的回调
def callback(future: Future):
    # Task 是 Future 的子类
    print(future.result())


task = loop.create_task(async_func(0.12))
task.add_done_callback(callback)
loop.run_until_complete(task)

# Future 是一种特殊的低层级 Awaitable 对象，表示一个异步操作的最终结果。
# ensure_future 函数将 Coroutine 或 Awaitable 对象包装成 Future
future = asyncio.ensure_future(async_func())
loop.run_until_complete(future)


# 并发运行任务

# asyncio.wait 函数将 Future 迭代器组合成一个协程
loop.run_until_complete(
    asyncio.wait([
        asyncio.ensure_future(async_func(0.1)),
        asyncio.ensure_future(async_func(0.2)),
        asyncio.ensure_future(async_func(0.3)),
    ])
)

# asyncio.gather 函数将传入的所有 Future 参数组合成一个迭代器
loop.run_until_complete(
    asyncio.gather(
        asyncio.ensure_future(async_func(0.1)),
        asyncio.ensure_future(async_func(0.2)),
        asyncio.ensure_future(async_func(0.3)),
    )
)

python 3.11 还提供了 TaskGroup

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(some_coro(...))
        task2 = tg.create_task(another_coro(...))
    print("Both tasks have completed now.")

←Back to home