[Python知识库]Python 协程Coroutine

?定义

协程，又称微线程，纤程。英文名Coroutine。通过 async/await 语法进行声明。需要 Python 3.7+

特点

协程的特点在于是一个线程执行，可以处理IO密集型程序的效率问题，但是处理CPU密集型不是它的长处，如要充分发挥CPU利用率可以结合多进程+协程。

?优势

最大的优势就是协程极高的执行效率。因为子程序切换不是线程切换，而是由程序自身控制，因此，没有线程切换的开销，和多线程比，线程数量越多，协程的性能优势就越明显。

第二大优势就是不需要多线程的锁机制，因为只有一个线程，也不存在同时写变量冲突，在协程中控制共享资源不加锁，只需要判断状态就好了，所以执行效率比多线程高很多。

因为协程是一个线程执行，那怎么利用多核CPU呢？最简单的方法是多进程+协程，既充分利用多核，又充分发挥协程的高效率，可获得极高的性能。

协程运行入口点：asyncio.run(coro, *, debug=False)

一：功能

1. 用来运行最高层级的入口点 "main()" 函数。
2. 此函数会运行传入的协程，负责管理 asyncio 事件循环，终结异步生成器，并关闭线程池。
3. 当有其他 asyncio 事件循环在同一线程中运行时，此函数不能被调用。
4. 如果 debug 为 True，事件循环将以调试模式运行。
5. 此函数总是会创建一个新的事件循环并在结束时关闭之。它应当被用作 asyncio 程序的主入口点，理想情况下应当只被调用一次。

挂起任务：coroutine asyncio.sleep(delay, result=None)

一：功能

挂起当前任务，以允许其他任务运行。

delay：阻塞指定的秒数。

如果指定了 result，则当协程完成时将其返回给调用者。

sleep() 总是会

将 delay 设为 0 将提供一个经优化的路径以允许其他任务运行。 这可供长期间运行的函数使用以避免在函数调用的全过程中阻塞事件循环。

?二：参数说明

1. delay：阻塞指定的秒数。将 delay 设为 0 将提供一个经优化的路径以允许其他任务运行。 这可供长期间运行的函数使用以避免在函数调用的全过程中阻塞事件循环。
2. result：如果指定了 result，则当协程完成时将其返回给调用者。

?三：示例

import asyncio

async def main():
     print('hello')
     # 把asyncio.sleep(1)看成是一个耗时1秒的IO操作，在此期间，主线程并未等待，
     # 而是去执行EventLoop中其他可以执行的coroutine了，因此可以实现并发执行。
     await asyncio.sleep(1)
     print('world')

# 运行协程main()
asyncio.run(main())

# Expected output:
# hello
# world

创建任务：asyncio.create_task(coro, *, name=None)

一：功能

用来并发运行作为 asyncio 任务的多个协程。

将 coro 协程封装为一个 Task 并调度其执行。返回 Task 对象。

二：参数说明

name 不为 None，它将使用 Task.set_name() 来设为任务的名称。

三：示例

import asyncio
import time

async def say_after(delay, what):
    print(what + f" started at {time.time()}")
    await asyncio.sleep(delay)
    print(what)

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')}")

asyncio.run(main())

# Output
"""
started at 17:14:32
hello started at 1635823295.3259559
world started at 1635823295.3259559
hello
world
finished at 17:14:34
"""

并发运行任务：asyncio.gather(）

awaitable asyncio.gather(*aws, return_exceptions=False)

一：功能

1. 并发运行 aws 序列中的可等待对象。
2. 如果 aws 中的某个可等待对象为协程，它将自动被作为一个任务调度。
3. 如果所有可等待对象都成功完成，结果将是一个由所有返回值聚合而成的列表。结果值的顺序与 aws 中可等待对象的顺序一致。

示例如下：

import asyncio
import time


async def cor(name):
    print(f'{name} start at {time.time()}')
    await asyncio.sleep(4)
    print(f'{name} end at {time.time()}')
    return name


async def main():
    try:
        print(f"started at {time.strftime('%X')}")
        task1 = asyncio.create_task(cor('task1'))
        task2 = asyncio.create_task(cor('task2'))
        task3 = asyncio.create_task(cor('task3'))
        task_gather = asyncio.gather(task1, task2, task3)

        print(await task_gather)

        print(f"finished at {time.strftime('%X')}")
    except asyncio.CancelledError:
        print(f"cancel at {time.strftime('%X')}")

asyncio.run(main())

# output:
"""
started at 17:33:14
task1 start at 1635931994.2334251
task2 start at 1635931994.2334251
task3 start at 1635931994.2334251
task1 end at 1635931998.2336676
task2 end at 1635931998.2336676
task3 end at 1635931998.2336676
['task1', 'task2', 'task3']
finished at 17:33:18
"""

?二：参数return_exceptions 为 False

return_exceptions 为 False (默认)，所引发的首个异常会立即抛出给等待 gather() 的任务。aws 序列中的其他可等待对象不会被取消并将继续运行。

示例如下：

import asyncio
import time


async def cor(name):
    print(f'{name} start at {time.time()}')
    await asyncio.sleep(4)
    print(f'{name} end at {time.time()}')
    return name


async def main():
    try:
        print(f"started at {time.strftime('%X')}")
        task1 = asyncio.create_task(cor('task1'))
        task2 = asyncio.create_task(cor('task2'))
        task3 = asyncio.create_task(cor('task3'))
        task1.cancel()
        task_gather = asyncio.gather(task1, task2, task3)

        print(await task_gather)

        print(f"finished at {time.strftime('%X')}")
    except asyncio.CancelledError:
        print(f"cancel at {time.strftime('%X')}")

asyncio.run(main())

# output:
"""
started at 17:30:40
task2 start at 1635931840.4336724
task3 start at 1635931840.434659
cancel at 17:30:40
"""

三：参数return_exceptions 为 True

参数return_exceptions 为 True时，异常会和成功的结果一样处理，并聚合至结果列表。

如果 aws 序列中的任一 Task 或 Future 对象 被取消，它将被当作引发了 CancelledError 一样处理 -- 在此情况下 gather() 调用 不会 被取消。这是为了防止一个已提交的 Task/Future 被取消导致其他 Tasks/Future 也被取消。

示例如下：

import asyncio
import time


async def cor(name):
    print(f'{name} start at {time.time()}')
    await asyncio.sleep(4)
    print(f'{name} end at {time.time()}')
    return name


async def main():
    print(f"started at {time.strftime('%X')}")
    task1 = asyncio.create_task(cor('task1'))
    task2 = asyncio.create_task(cor('task2'))
    task3 = asyncio.create_task(cor('task3'))
    task1.cancel()
    task_gather = asyncio.gather(task1, task2, task3, return_exceptions=True)

    print(await task_gather)

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

asyncio.run(main())

# output:
"""
started at 17:28:32
task2 start at 1635931712.9419875
task3 start at 1635931712.9419875
task2 end at 1635931716.9421253
task3 end at 1635931716.9421253
[CancelledError(), 'task2', 'task3']
finished at 17:28:36
"""

?四：gather()被取消时

如果 gather() 被取消，所有被提交(尚未完成) 的可等待对象也会被取消。

示例如下：

import asyncio
import time


async def cor(name):
    print(f'{name} start at {time.time()}')
    await asyncio.sleep(4)
    print(f'{name} end at {time.time()}')
    return name


async def main():
    try:
        print(f"started at {time.strftime('%X')}")
        task1 = asyncio.create_task(cor('task1'))
        task2 = asyncio.create_task(cor('task2'))
        task3 = asyncio.create_task(cor('task3'))
        gather1 = asyncio.gather(task1, task2, task3)
        gather1.cancel()

        task4 = asyncio.create_task(cor('task4'))
        task5 = asyncio.create_task(cor('task5'))
        gather2 = asyncio.gather(task4, task5)
        L = await asyncio.gather(gather1, gather2, return_exceptions=True)
        print(L)

        print(f"finished at {time.strftime('%X')}")
    except asyncio.CancelledError:
        print(f"cancel at {time.strftime('%X')}")
    except asyncio.TimeoutError:
        print(f"time out at {time.strftime('%X')}")

asyncio.run(main())

# output:
"""
started at 17:22:49
task4 start at 1635931369.0492625
task5 start at 1635931369.0492625
task4 end at 1635931373.0496387
task5 end at 1635931373.0496387
[CancelledError(), ['task4', 'task5']]
finished at 17:22:53
"""

# 执行结果中可以看出，gather1取消时，task1、task2、task3都被取消了

?coroutine asyncio.wait_for(aw, timeout)

一：功能

等待 aw执行完成，指定 timeout 秒数后超时。如果 aw 是一个协程，它将自动被作为任务调度。

二：参数 timeout

1. 为 None时，则等待任务完成
2. 非None，为 float 或 int 型数值时，表示等待的秒数。aw执行超过了timeout的设定值时，任务aw将会被取消，并引发 asyncio.TimeoutError。如果取消期间发生了异常，异常将会被抛出。
3. 示例如下：

import asyncio
import time


async def cor(name):
    print(f'{name} start at {time.time()}')
    await asyncio.sleep(4)
    print(f'{name} end at {time.time()}')


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

        await asyncio.wait_for(cor('task1'), timeout=1)
        
        print(f"finished at {time.strftime('%X')}")
    except asyncio.TimeoutError:
        print(f"time out at {time.strftime('%X')}")

asyncio.run(main())

# output:
"""
started at 14:06:03
task1 start at 1635919563.2111363
time out at 14:06:04
"""

?三：要避免任务取消，可以配合 asyncio.shield()

需要注意的是，asyncio.shield()避免任务取消的功能，只有在执行多个任务时，才生效。

示例如下，

import asyncio
import time


async def cor(name):
    print(f'{name} start at {time.time()}')
    await asyncio.sleep(4)
    print(f'{name} end at {time.time()}')


async def main():
    print(f"started at {time.strftime('%X')}")
    
    # asyncio.wait_for(),等待cor('task1')执行完成，超过设定的timeout=1s时，取消任务cor('task1')
    # asyncio.shield()保护cor('task1')不被取消，使其执行完成
    cor_wait = asyncio.wait_for(asyncio.shield(cor('task1')), timeout=1)
    task1 = asyncio.create_task(cor_wait)
    task2 = asyncio.create_task(cor('task2'))

    await asyncio.gather(task1, task2, return_exceptions=True)

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

asyncio.run(main())

# output:
"""
started at 13:38:19
task1 start at 1635917899.778396
task2 start at 1635917899.778396
task1 end at 1635917903.7799335
task2 end at 1635917903.7799335
finished at 13:38:23
"""

?四：如果等待被取消，则aw指定的对象也会被取消

import asyncio
import time


async def cancel_after(task, when):
    print(f'cancel at {time.time()}')
    await asyncio.sleep(when)
    task.cancel()


async def cor(name):
    print(f'{name} start at {time.time()}')
    await asyncio.sleep(4)
    print(f'{name} end at {time.time()}')


async def main():
    try:
        print(f"started at {time.strftime('%X')}")
        wait_task = asyncio.create_task(asyncio.wait_for(cor('task1'), timeout=None))
        await cancel_after(wait_task, 1)

        print(f"finished at {time.strftime('%X')}")
    except asyncio.TimeoutError:
        print(f"time out at {time.strftime('%X')}")

asyncio.run(main())

# output:
"""
started at 14:22:00
cancel at 1635920520.3298035
task1 start at 1635920520.3298035
finished at 14:22:01

"""


# 从执行结果中可以看出，cor('task1')未执行完成，即取消了后续操作

awaitable asyncio.shield(aw)

保护一个可等待对象防止其被取消。如果 aw 是一个协程，它将自动被作为任务调度。

如果

res = await shield(something())
相当于:
res = await something()

不同之处在于, shield(something()),如果包含它的协程被取消，在 something() 中运行的任务不会被取消。

注意：程序中，只有一个等待对象执行，shield()函数不会生效。多个等待对象执行时，才会生效。参考示例

从 something() 的角度看来，取消操作并没有发生。然而其调用者已被取消，因此 "await" 表达式仍然会引发 CancelledError。

如果通过其他方式取消 something() (例如在其内部操作) 则 shield() 也会取消。

如果希望完全忽略取消操作 (不推荐) ，则 shield() 函数需要配合一个 try/except 代码段，如下所示:

try:
? ? res = await asyncio.shield(something())
except CancelledError:
? ? res = None

?asyncio.shield示例一：单个协程，不生效

import asyncio
import time


async def cor(name):
    print(f'{name} start at {time.time()}')
    await asyncio.sleep(4)
    print(f'{name} end at {time.time()}')


async def main():
    print(f"started at {time.strftime('%X')}")
     # asyncio.wait_for(),等待cor('task1')执行完成，超过设定的timeout=1s时，取消任务cor('task1')
    # asyncio.shield()保护cor('task1')不被取消，使其执行完成
    cor_wait = asyncio.wait_for(asyncio.shield(cor('task1')), timeout=1)
    task1 = asyncio.create_task(cor_wait)
    await asyncio.gather(task1, return_exceptions=True)
    print(f"finished at {time.strftime('%X')}")

asyncio.run(main())

# output:
"""
started at 13:36:18
task1 start at 1635917778.904956
finished at 13:36:19
"""

从运行结果中，可以看到，任务cor('task1')没有执行完成，asyncio.shield()没有生效。

??asyncio.shield示例二：多个协程，生效

import asyncio
import time


async def cor(name):
    print(f'{name} start at {time.time()}')
    await asyncio.sleep(4)
    print(f'{name} end at {time.time()}')


async def main():
    print(f"started at {time.strftime('%X')}")
    
    # asyncio.wait_for(),等待cor('task1')执行完成，超过设定的timeout=1s时，取消任务cor('task1')
    # asyncio.shield()保护cor('task1')不被取消，使其执行完成
    cor_wait = asyncio.wait_for(asyncio.shield(cor('task1')), timeout=1)
    task1 = asyncio.create_task(cor_wait)
    task2 = asyncio.create_task(cor('task2'))

    await asyncio.gather(task1, task2, return_exceptions=True)

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

asyncio.run(main())

# output:
"""
started at 13:38:19
task1 start at 1635917899.778396
task2 start at 1635917899.778396
task1 end at 1635917903.7799335
task2 end at 1635917903.7799335
finished at 13:38:23
"""

运行结果中看出，asyncio.shield()使cor('task1')执行完成

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

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

aws 可迭代对象必须不为空。

返回两个 Task/Future 集合: (done, pending)。

如指定 timeout (float 或 int 类型) 则它将被用于控制返回之前等待的最长秒数。

请注意此函数不会引发 asyncio.TimeoutError。当超时发生时，未完成的 Future 或 Task 将在指定秒数后被返回。

return_when 指定此函数应在何时返回。它必须为以下常数之一:

1. FIRST_COMPLETED：函数将在任意可等待对象结束或取消时返回。
2. FIRST_EXCEPTION：函数将在任意可等待对象因引发异常而结束时返回。当没有引发任何异常时它就相当于 ALL_COMPLETED。
3. ALL_COMPLETED：函数将在所有可等待对象结束或取消时返回。

与 wait_for() 不同，wait() 在超时发生时不会取消可等待对象。

coroutine asyncio.to_thread(func, /, *args, **kwargs)

在不同的线程中异步地运行函数 func。

向此函数提供的任何 *args 和 **kwargs 会被直接传给 func。

并且，当前 contextvars.Context 会被传播，允许在不同的线程中访问来自事件循环的上下文变量。

返回一个可被等待以获取 func 的最终结果的协程。

这个协程函数主要是用于执行在其他情况下会阻塞事件循环的 IO 密集型函数/方法。

def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    # Note that time.sleep() can be replaced with any blocking
    # IO-bound operation, such as file operations.
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

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

    await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1))

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


asyncio.run(main())

# Expected output:
#
# started main at 19:50:53
# start blocking_io at 19:50:53
# blocking_io complete at 19:50:54
# finished main at 19:50:54

注解

?

由于?GIL?的存在，asyncio.to_thread()?通常只能被用来将 IO 密集型函数变为非阻塞的。 但是，对于会释放 GIL 的扩展模块或无此限制的替代性 Python 实现来说，asyncio.to_thread()?也可被用于 CPU 密集型函数。

asyncio.run_coroutine_threadsafe(coro, loop)

向指定事件循环提交一个协程。（线程安全）

返回一个 concurrent.futures.Future 以等待来自其他 OS 线程的结果。

此函数应该从另一个 OS 线程中调用，而非事件循环运行所在线程。

不同与其他 asyncio 函数，此函数要求显式地传入 loop 参数。

# Create a coroutine
coro = asyncio.sleep(1, result=3)

# Submit the coroutine to a given loop
future = asyncio.run_coroutine_threadsafe(coro, loop)

# Wait for the result with an optional timeout argument
try:
    result = future.result(timeout)
except concurrent.futures.TimeoutError:
    print('The coroutine took too long, cancelling the task...')
    future.cancel()
except Exception as exc:
    print(f'The coroutine raised an exception: {exc!r}')
else:
    print(f'The coroutine returned: {result!r}')

asyncio.current_task(loop=None)

返回当前运行的 Task 实例，如果没有正在运行的任务则返回 None。

如果 loop 为 None 则会使用 get_running_loop() 获取当前事件循环。

asyncio.all_tasks(loop=None)

返回事件循环所运行的未完成的 Task 对象的集合。

如果 loop 为 None，则会使用 get_running_loop() 获取当前事件循环。

可等待对象

如果一个对象可以在 await 语句中使用，那么它就是 可等待 对象。许多 asyncio API 都被设计为接受可等待对象。

可等待 对象有三种主要类型: 协程, 任务 和 Future.

任务

任务 被用来“并行的”调度协程

当一个协程通过 asyncio.create_task() 等函数被封装为一个 任务，该协程会被自动调度执行:

asyncio.create_task(coro, *, name=None)

Futures

Future 是一种特殊的 低层级 可等待对象，表示一个异步操作的 最终结果。

当一个 Future 对象 被等待，这意味着协程将保持等待直到该 Future 对象在其他地方操作完毕。

在 asyncio 中需要 Future 对象以便允许通过 async/await 使用基于回调的代码。

通常情况下 没有必要 在应用层级的代码中创建 Future 对象。

Future 对象有时会由库和某些 asyncio API 暴露给用户，用作可等待对象:

一个很好的返回对象的低层级函数的示例是 loop.run_in_executor()。

示例

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import threading
import asyncio

async def hello():
    print('Hello world! (%s)' % threading.currentThread())
    await 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()