异步 python itertools 链接多个生成器

Question

更新的清晰度问题：假设我有 2 个处理生成器函数：def gen1(): # just for examples,  yield 1   # yields actually carry   yield 2   # different computation weight   yield 3   # in my casedef gen2():  yield 4  yield 5  yield 6我可以用 itertools 链接它们from itertools import chainmix = chain(gen1(), gen2())然后我可以用它创建另一个生成器函数对象，def mix_yield():   for item in mix:      yield item或者只是如果我只是想next(mix)，它就在那里。我的问题是，如何在异步代码中执行等效操作？因为我需要它：以收益方式返回（一一），或使用next迭代器首先解决最快的产量（异步）

Answer 1

我遇到了这个答案，并查看了 aiostream 库。这是我想出的用于合并多个异步生成器的代码。它不使用任何库。

async def merge_generators(gens:Set[AsyncGenerator[Any, None]]) -> AsyncGenerator[Any, None]:

    pending = gens.copy()

    pending_tasks = { asyncio.ensure_future(g.__anext__()): g for g in pending }

    while len(pending_tasks) > 0:

        done, _ = await asyncio.wait(pending_tasks.keys(), return_when="FIRST_COMPLETED")

        for d in done:

            try:

                result = d.result()

                yield result

                dg = pending_tasks[d]

                pending_tasks[asyncio.ensure_future(dg.__anext__())] = dg

            except StopAsyncIteration as sai:

                print("Exception in getting result", sai)

            finally:

                del pending_tasks[d]

希望这对您有所帮助，如果有任何错误，请告诉我。