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

Question

我试图正确理解和实现两个并发运行的任务对象使用Python 3相对较新的异步模块。

简而言之，asyncio似乎被设计用于处理异步进程和事件循环上的并发Task执行。它提倡使用wait（应用于异步函数）作为无回调的方式来等待和使用结果，而不阻塞事件循环。（期货和回调仍然是可行的选择。）

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

我喜欢能够使用并发任务而不需要使用Executor类的想法，但我还没有发现太多关于实现的细节。

我现在就是这样做的：

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()

在试图同时运行两个循环任务的情况下，我注意到，除非任务有一个内部的wait表达式，否则它将卡在而循环中，有效地阻止其他任务运行（很像正常的>而循环）。然而，一旦任务必须等待，它们似乎就会并发运行，没有问题。

因此，wait语句似乎为事件循环提供了在任务之间来回切换的立足点，从而产生了并发效果。

示例输出与内部等待：

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

没有内部等待的输出示例：

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

这个实现是否传递了asyncio中并发循环任务的适当示例？

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




Answer 1

您不一定需要从x获得收益来控制事件循环。

在您的示例中，我认为正确的方法是执行一个收益无，或者等效地执行一个简单的收益，而不是从asyncio.sleep（0.001）生成一个：

import asyncio

@asyncio.coroutine
def say_boo():
  i = 0
  while True:
    yield None
    print("...boo {0}".format(i))
    i += 1

@asyncio.coroutine
def say_baa():
  i = 0
  while True:
    yield
    print("...baa {0}".format(i))
    i += 1

boo_task = asyncio.async(say_boo())
baa_task = asyncio.async(say_baa())

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

协同程序只是普通的老Python生成器。在内部，asyncio事件循环保存这些生成器的记录，并在一个永无止境的循环中逐个调用gen.send（）。无论何时产生，对gen.send（）的调用都将完成，循环可以继续。（我正在简化它；看看周围https://hg.python.org/cpython/file/3.4/Lib/asyncio/tasks.py#l265 （对于实际代码）

也就是说，如果你需要在不共享数据的情况下进行CPU密集型计算，我仍然会走run_in_executor路线。

Answer 2

是的，在事件循环中运行的任何协同程序都会阻止其他协同程序和任务运行，除非

1. 使用从或wait调用另一个协程（如果使用Python 3.5）。
2. 返回。

这是因为asyncio是单线程的；事件循环运行的唯一方法是不让其他协同程序积极执行。使用yield from/await暂时挂起协同程序，使事件循环有机会工作。

您的示例代码很好，但是在许多情况下，您可能不希望长时间运行的代码在事件循环中不执行异步I/O。在这种情况下，使用asyncio.loop.run_in_executor在后台线程或进程中运行代码通常更有意义。如果您需要执行一些对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 = loop.run_in_executor(executor, say_boo)
    baa = loop.run_in_executor(executor, say_baa)

    loop.run_forever()