简短概述:正确的方法
异步生成器不适合并行消费。请看下面的解释。作为一个适当的解决方法,使用asyncio.Queue
来实现生产者和消费者之间的通信:
queue = asyncio.Queue()
async def producer():
for item in range(5):
await asyncio.sleep(random.random())
print('item fetched:', item)
await queue.put(item)
async def consumer():
while True:
item = await queue.get()
await asyncio.sleep(random.random())
print('item processed:', item)
await asyncio.gather(producer(), consumer(), consumer())
上面的代码片段适用于无限流的项目:例如,一个永远运行并为客户端提供请求的Web服务器。但是,如果我们需要处理有限数量的项目呢?消费者应该如何知道何时停止?
这值得在Stack Overflow上提出另一个问题来涵盖所有的替代方案,但最简单的选择是使用下面描述的sentinel
方法。
Sentinel: 有限数据流方法
引入一个sentinel = object()
。当从外部数据源获取并将所有项目放入队列时,producer
必须向队列中推送与您拥有的consumer
数量相同的sentinel
。一旦consumer
获取了sentinel
,它就知道应该停止:if item is sentinel: break
退出循环。
sentinel = object()
consumers_count = 2
async def producer():
...
if new_item is None:
for _ in range(consumers_count):
await queue.put(sentinel)
async def consumer():
while True:
...
if item is sentinel:
break
await asyncio.gather(
producer(),
*(consumer() for _ in range(consumers_count)),
)
TL;DR [2]: 一个不太干净的解决方法
由于您需要不更改异步生成器方法,因此这里提供了一种基于 asyncgen 的替代方案。为了以一种简单但不太干净的方式解决此问题,您可以使用锁来包装源异步生成器:
async def with_lock(agen, lock: asyncio.Lock):
while True:
async with lock:
try:
item = await anext(agen)
except StopAsyncIteration:
break
yield item
lock = asyncio.Lock()
await asyncio.gather(
anext(with_lock(datas, lock)),
anext(with_lock(datas, lock)),
...
)
这将确保每次只有一个消费者从生成器中等待获取项目。当此消费者等待时,其他消费者正在执行,因此并行性不会丢失。
重要提示!在上述方法中,不要将yield await anext(agen)
作为单个表达式放在lock
下:您的包装生成器将挂起(在yield
上),并且lock
未释放,没有其他消费者能够并行消费另一个项目。即仅使用锁包装anext
调用,但不要在锁定部分中yield
。
使用async for
的大致等效代码(看起来更智能):
async def with_lock(agen, lock: asyncio.Lock):
await lock.acquire()
async for item in agen:
lock.release()
yield item
await lock.acquire()
lock.release()
然而,这段代码仅处理异步生成器的anext
方法。而生成器API还包括aclose
和athrow
方法。下面进行解释。
虽然您可以将对这些方法的支持添加到with_lock
函数中,但我建议要么子类化一个生成器并在内部处理锁支持,要么更好地使用上面的基于Queue
的方法。
参见contextlib.aclosing
以获取一些灵感。
解释
同步和异步生成器都有一个特殊属性:.gi_running
(用于常规生成器)和.ag_running
(用于异步生成器)。 您可以通过在生成器上执行dir
来发现它们:
>>> dir((i for i in range(0))
[..., 'gi_running', ...]
当生成器的.__next__
或.__anext__
方法被执行时(next(...)
和anext(...)
只是语法糖),它们被设置为True
。
这可以防止在一个生成器上重新执行next(...)
,当同一生成器上的另一个next(...)
调用已经在执行时:如果运行标志为True
,则会引发异常(对于同步生成器,它会引发ValueError: generator already executing
)。
因此,回到您的示例,当您运行await anext(datas)
(通过asyncio.gather
)时,会发生以下情况:
datas.ag_running
被设置为 True
。
- 执行流程进入
datas.__anext__
方法。
- 一旦在
__anext__
方法内部达到了一个内部的 await
语句(例如你的 await asyncio.sleep(1)
),asyncio
的循环将切换到另一个消费者。
- 现在,另一个消费者也尝试调用
await anext(datas)
,但由于 datas.ag_running
标志仍然被设置为 True
,这会导致 RuntimeError
。
为什么需要这个标志?
生成器的执行可以暂停和恢复。但只能在 yield
语句处。因此,如果生成器在内部的 await
语句处暂停,它就无法“恢复”,因为它的状态不允许。
这就是为什么对生成器进行并行的 next
/anext
调用会引发异常的原因:它还没有准备好被恢复,它已经在运行中了。
athrow
和aclose
生成器的API(同步和异步)不仅包括用于迭代的send
/asend
方法,还包括:
close
/aclose
用于在退出或异常时释放生成器分配的资源(例如数据库连接)
throw
/athrow
用于通知生成器必须处理异常。
aclose
和athrow
也是异步方法。这意味着如果两个消费者尝试并行关闭/抛出底层生成器,则会遇到相同的问题,因为在关闭(或处理异常)生成器时,它又被关闭(抛出异常)。
同步生成器示例
虽然这是异步生成器的常见情况,但对于同步生成器来说,重现它并不那么简单,因为同步next(...)
调用很少被中断。
中断同步生成器的一种方法是运行具有多个消费者的多线程代码(在并行线程中运行),从单个生成器读取。 在这种情况下,当生成器的代码在执行next
调用时被中断时,所有其他消费者并行尝试调用next
将导致异常。
另一种实现此目的的方法是通过自消耗生成器在生成器相关PEP#255中演示:
>>> def g():
... i = next(me)
... yield i
...
>>> me = g()
>>> next(me)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<stdin>", line 2, in g
ValueError: generator already executing
当外部调用next(me)
时,它会将me.gi_running
设置为True
,然后执行生成器函数代码。随后的内部next(me)
调用会导致ValueError
。
结论
生成器(特别是异步)最适合由单个读取器使用。多个消费者的支持很困难,因为需要修补所有生成器方法的行为,因此不鼓励这样做。
get_data
执行了asyncio.sleep
模拟的任何操作并返回其结果,那么你只需收集get_data
五次(或实际数字是多少次)即可。 - dirn