更新的清晰度問題：假設(shè)我有 2 個處理生成器函數(shù)：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())然后我可以用它創(chuàng)建另一個生成器函數(shù)對象，def mix_yield(): for item in mix: yield item或者只是如果我只是想next(mix)，它就在那里。我的問題是，如何在異步代碼中執(zhí)行等效操作？因為我需要它：以收益方式返回（一一），或使用next迭代器首先解決最快的產(chǎn)量（異步）
2 回答
守候你守候我
TA貢獻(xiàn)1802條經(jīng)驗 獲得超10個贊
我遇到了這個答案，并查看了 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]
希望這對您有所幫助，如果有任何錯誤，請告訴我。
添加回答
舉報
0/150
提交
取消