协程与任务

Posted 2022-08-09 哦...

在开始之前先明确两点：

1. 使用async def定义协程函数/协程是目前推荐的方式。
2. asyncio是一个帮助开发者进行异步编程的工具库。

1. 协程

Python3.5之后应通过 async/await 语法来声明协程函数/协程。

>>> import asyncio

>>> async def main():
...     print('hello')
...     await asyncio.sleep(1)
...     print('world')

>>> asyncio.run(main())
hello
world

就像有yield的函数就不是函数，“调用”函数时获得的是一个生成器对象。用async def定义的函数称为协程函数，“调用”协程函数时获得的是一个协程对象（coroutine object）。

>>> main()
<coroutine object main at 0x1053bb7c8>

要想让协程运行获得结果，必须为协程提供事件总线。asyncio提供了三种方式方便协程的运行

第一种方式：asyncio.run(协程）

>>> asyncio.run(main())
hello
world

 第二种方式：await 协程

import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    print(f"started at time.strftime('%X')")

    await say_after(1, 'hello')
    await say_after(2, 'world')

    print(f"finished at time.strftime('%X')")

asyncio.run(main())

await有一定的局限性，await必须写在一个async def中，也就是要写在另一个协程中。

例子中main()是一个协程，say_after是一个协程。用第一种方式执行main()协程时，用第二种方式在main()协程内部完成了两个say_after()协程的执行。

第三种方式：task = asyncio.create_task(协程）将协程包装为Task类型的对象。Task类型是Future（期物）类型的子类。

async def main():
    task1 = asyncio.create_task(say_after(1, 'hello'))

    task2 = asyncio.create_task(say_after(2, 'world'))

    print(f"started at time.strftime('%X')")

    # Wait until both tasks are completed (should take around 2 seconds.)
    await task1
    await task2

    print(f"finished at time.strftime('%X')")

修改main()，利用asyncio提供的工具函数create_task()将协程say_after()包装为Task对象。但是Task对象的执行也得用await的方式执行。

task1、task2任务会在get_running_loop()返回的事件总线中执行，如果当前线程没有在运行的事件总线则会引发RuntimeError。

所以，到底什么是可以写在await后面的呢？

2. 可等待的(Awaitable)

如果一个对象可以在await语句中使用，那么它就是 可等待 对象。许多 asyncio API 都被设计为接受可等待对象，包括协程(Coroutine), 任务(Task) 和 期物(Future)对象。

期物相较协程、任务来说是一种特殊的低层级可等待对象，表示一个异步操作的最终结果。

期物最典型的应用场景是通过asyncio提供的run_in_executor函数，该函数将一个非异步阻塞函数通过扔到线程池中执行，同时将该函数包装为一个Future（期物）对象返回。

注意，在异步执行上下文中，绝对不能出现非异步阻塞函数，因为它会阻塞线程，进而阻塞事件总线对总线上任务的执行。

3. 并发执行批量任务

awaitable asyncio.gather(*aws, return_exceptions=False) 并发 运行aws序列中的可等待对象。注意传参的方式，是依次传入awaitable1,awaitable2,awaitable3...

asyncio的gather与线程池的map有异曲同工之妙。只不过一个是以单线程异步方式执行任务（期物对象的执行另当别论），一个是多线程执行任务，最后执行的结果都以序列作为容器，以任务添加顺序作为下标，有序呈现。

import asyncio

async def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print(f"Task name: Compute factorial(number), currently i=i...")
        await asyncio.sleep(1)
        f *= i
    print(f"Task name: factorial(number) = f")
    return f

async def main():
    # Schedule three calls *concurrently*:
    L = await asyncio.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    )
    print(L)

asyncio.run(main())

逢await时会将执行上下文给予事件总线。

# Expected output:
#
#     Task A: Compute factorial(2), currently i=2...
#     Task B: Compute factorial(3), currently i=2...
#     Task C: Compute factorial(4), currently i=2...
#     Task A: factorial(2) = 2
#     Task B: Compute factorial(3), currently i=3...
#     Task C: Compute factorial(4), currently i=3...
#     Task B: factorial(3) = 6
#     Task C: Compute factorial(4), currently i=4...
#     Task C: factorial(4) = 24
#     [2, 6, 24]

4.超时与等待

coroutine asyncio.wait(aws, *, timeout=None, return_when=ALL_COMPLETED)

并发地运行参数中的可等待对象们 并进入阻塞状态直到满足 return_when 所指定的条件。

aws 可迭代对象必须不为空。返回两个 Task/Future 集合: (done, pending)。

done, pending = await asyncio.wait(aws)

 coroutine asyncio.wait_for(aw, timeout)等待aw 完成，指定 timeout 秒数后超时。

如果 aw 是一个协程，它将自动被作为任务调度。如果发生超时，任务将取消并引发asyncio.TimeOutError。

以上部大部分案例与内容摘自Python官网，有编辑。