如何使用python的asyncio模块正确创建和运行并发任务？

Question

我正在尝试Task使用Python
3的相对较新的asyncio模块正确理解和实现两个同时运行的对象。

简而言之，asyncio似乎旨在处理Task事件循环中的异步过程和并发执行。它促进了await（应用于异步函数中）作为无回调方式等待和使用结果的使用，而不会阻塞事件循环。（期货和回调仍然是可行的选择。）

它还提供了asyncio.Task()该类，该类Future
是用于包装协程的专门子类。最好通过使用该asyncio.ensure_future()方法来调用。异步任务的预期用途是允许独立运行的任务与同一事件循环中的其他任务“同时”运行。我的理解是Tasks连接到事件循环，然后该循环自动继续驱动await语句之间的协程。

我喜欢无需使用任何一个Executor类就可以使用并发Task的想法，但是我并没有在实现上做过多的阐述。

这是我目前正在做的事情：

import asyncio

print('running async test')

async def say_boo():
    i = 0
    while True:
        await asyncio.sleep(0)
        print('...boo {0}'.format(i))
        i += 1

async def say_baa():
    i = 0
    while True:
        await asyncio.sleep(0)
        print('...baa {0}'.format(i))
        i += 1

# wrap in Task object
# -> automatically attaches to event loop and executes
boo = asyncio.ensure_future(say_boo())
baa = asyncio.ensure_future(say_baa())

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

在尝试同时运行两个循环任务的情况下，我注意到，除非该任务具有内部await表达式，否则它将卡在while循环中，从而有效地阻止了其他任务的运行（就像正常while循环一样）。但是，一旦任务必须（a）等待，它们似乎可以并发运行而没有问题。

因此，这些await语句似乎为事件循环提供了一个立足点，可在任务之间来回切换，从而带来并发效果。

内部输出示例await：

running async test
...boo 0
...baa 0
...boo 1
...baa 1
...boo 2
...baa 2

不带 内部输出的示例await：

...boo 0
...boo 1
...boo 2
...boo 3
...boo 4

此实现是否通过了并发循环Task的“适当”示例asyncio？

唯一可行的方法是Task提供一个阻塞点（await表达式）以便事件循环处理多个任务是否正确？

Answer 1

是的，事件循环中运行的任何协程将阻止其他协程和任务运行，除非它

1. 使用yield from或await（如果使用Python 3.5+）调用另一个协程。
2. 退货。

这是因为asyncio是单线程的。事件循环运行的唯一方法是没有其他协程正在积极执行。使用yield from/await暂时暂停协程，使事件循环有工作的机会。

您的示例代码很好，但是在很多情况下，您可能不希望开始时没有在事件循环内运行不执行异步I /
O的长时间运行的代码。在这些情况下，通常更适合asyncio.loop.run_in_executor在后台线程或进程中运行代码。ProcessPoolExecutor如果您的任务是CPU限制的，则是更好的选择；如果您ThreadPoolExecutor需要执行一些asyncio不友好的I
/ O ，则将使用它。

例如，您的两个循环完全受CPU限制，并且不共享任何状态，因此，最好的性能来自ProcessPoolExecutor于跨CPU并行运行每个循环：

import asyncio
from concurrent.futures import ProcessPoolExecutor

print('running async test')

def say_boo():
    i = 0
    while True:
        print('...boo {0}'.format(i))
        i += 1


def say_baa():
    i = 0
    while True:
        print('...baa {0}'.format(i))
        i += 1

if __name__ == "__main__":
    executor = ProcessPoolExecutor(2)
    loop = asyncio.get_event_loop()
    boo = asyncio.create_task(loop.run_in_executor(executor, say_boo))
    baa = asyncio.create_task(loop.run_in_executor(executor, say_baa))

    loop.run_forever()