TL;DR：正确的方式
异步生成器非常不适合并行使用。请参阅下面的解释。作为一种适当的解决方法，使用asyncio.Queue 进行生产者和消费者之间的通信：
queue = asyncio.Queue()
async def producer():
for item in range(5):
await asyncio.sleep(random.random()) # imitate async fetching
print('item fetched:', item)
await queue.put(item)
async def consumer():
while True:
item = await queue.get()
await asyncio.sleep(random.random()) # imitate async processing
print('item processed:', item)
await asyncio.gather(producer(), consumer(), consumer())
上面的代码 sn-p 适用于无穷项目流：例如，一个 Web 服务器，它永远运行，为来自客户端的请求提供服务。但是如果我们需要处理有限数量的项目呢？consumers 应该如何知道何时停止？
这值得在 Stack Overflow 上提出另一个问题以涵盖所有替代方案，但最简单的选择是sentinel 方法，如下所述。
Sentinel：有限数据流方法
介绍一个sentinel = object()。当从外部数据源获取所有项目并将其放入队列时，producer 必须将与您拥有的 consumers 一样多的 sentinels 推送到队列中。一旦consumer 获取sentinel，它就知道它应该停止：if item is sentinel: break from loop。
sentinel = object()
consumers_count = 2
async def producer():
... # the same code as above
if new_item is None: # if no new data
for _ in range(consumers_count):
await queue.put(sentinel)
async def consumer():
while True:
... # the same code as above
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: # only one consumer is allowed to read
try:
yield await anext(agen)
except StopAsyncIteration:
break
lock = asyncio.Lock() # a common lock for all consumers
await asyncio.gather(
# every consumer must have its own "wrapped" generator
anext(with_lock(datas, lock)),
anext(with_lock(datas, lock)),
...
)
这将确保只有一个消费者等待来自生成器的项目一次.在此消费者等待时，其他消费者正在执行，因此不会丢失并行化。
与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__ 方法时，它们被设置为True（next(...) 和anext(...) 只是它们的语法糖）。
这可以防止在同一生成器上的另一个 next(...) 调用已经在执行时在生成器上重新执行 next(...)：如果运行标志是 True，则会引发异常（对于同步生成器，它会引发 ValueError: generator already executing ）。
因此，回到您的示例，当您运行await anext(datas)（通过asyncio.gather）时，会发生以下情况：
-
datas.ag_running 设置为 True。
- 执行流程进入
datas.__anext__ 方法。
- 一旦在
__anext__ 方法（在您的情况下为await asyncio.sleep(1)）内部到达await 语句，asyncio 的循环将切换到另一个使用者。
- 现在，另一个消费者也尝试调用
await anext(datas)，但由于datas.ag_running 标志仍设置为True，这将导致RuntimeError。
为什么需要这个标志？
生成器的执行可以暂停和恢复。但仅限于yield 声明。因此，如果生成器在内部 await 语句处暂停，它不能被“恢复”，因为它的状态不允许它。
这就是为什么并行调用生成器 next/anext 会引发异常：它还没有准备好恢复，它已经开始.
athrow 和 aclose
Generators 的 API（同步和异步）不仅包括 send/asend 迭代方法，还包括：
-
close/aclose 在退出或异常时释放生成器分配的资源（例如数据库连接）
- 和
throw/athrow 通知生成器它必须处理异常。
aclose 和 athrow 也是异步方法。这意味着如果两个消费者尝试并行关闭/抛出底层生成器，您将遇到相同的问题，因为生成器将在关闭（或处理异常）时再次关闭（抛出异常）。
同步生成器示例
尽管这是异步生成器的常见情况，但为同步生成器复制它并不是那么天真，因为同步next(...) 调用很少被中断。
中断同步生成器的一种方法是运行多线程代码，其中多个使用者（以并行线程运行）从单个生成器中读取数据。在这种情况下，当生成器的代码在执行next 调用时被中断，所有其他消费者并行尝试调用next 将导致异常。
实现此目的的另一种方法在generators-related 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。
结论
生成器（尤其是异步）在由单个阅读器使用时效果最佳。多个消费者支持很难，因为需要修补所有生成器方法的行为，因此不鼓励。