如何处理协同程序函数的多个结果？

Question

我有一些生成器做一些搜索的东西,我使用另一个生成器包装它们:

def searching_stuff_1():
    # searching
    yield 1
    # and searching
    yield 2
    yield 3

def searching_stuff_2():
    yield 4
    yield 5


def gen():
    yield from searching_stuff_1()
    yield from searching_stuff_2()

for result in gen():
    print(result)

所以现在我想知道如何将其重写为异步版本,这可以在searching_stuff_1和searching_stuff_2中产生多个值.

我在努力:

import asyncio

async def searching_stuff_1():
    f = asyncio.Future()
    result = []
    await asyncio.sleep(1)
    #searching
    result.append(1)
    #searching
    result.append(2)
    result.append(3)
    f.set_result(result)
    return f

async def searching_stuff_2():
    f = asyncio.Future()
    result = []
    await asyncio.sleep(1)
    result.append(4)
    result.append(5)
    f.set_result(result)
    return f

async def producer():
    coros = [searching_stuff_1(), searching_stuff_2()]
    for future in asyncio.as_completed(coros):
        yield await future

async def consumer(xs):
    async for future in xs:
        for r in future.result():
            print(r)
loop = asyncio.get_event_loop()
loop.run_until_complete(consumer(producer()))
loop.close()

但是,在这个版本中,我必须将所有结果附加到列表中并将其包装在Future实例中.我想知道是否有更好的方法来处理协程函数的多个结果.我有可能仍然能够获得这些数字吗？

Answer 1

是的，您仍然可以在异步版本中生成这些数字，这是异步生成器，您可以使用 async for (PEP492)和异步推导式(PEP530)，像这样，从第一个示例重写。虽然这需要 python 版本高于或等于 3.6

import asyncio


async def searching_stuff_1():
    # searching
    yield 1
    # and searching
    yield 2
    yield 3


async def searching_stuff_2():
    yield 4
    yield 5


async def gen():
    async for i in searching_stuff_1():
        yield i
    async for i in searching_stuff_2():
        yield i


async def gen_all():
    return [i async for i in gen()]

if __name__ == "__main__":
    result = asyncio.get_event_loop().run_until_complete(gen_all())
    print(result)

如果希望运行两个异步生成器，您可以使用asyncio.gather.
但由于asyncio.gather仅以异步方式收集协程的结果，因此您需要async def gen2在调用 asyncio.gather 之前分别组合异步生成器的每个生成结果，

async def gen2(coro):
    return [i async for i in coro()]


# combine two async
async def gen_all2():
    return list(chain.from_iterable(await gather(gen2(searching_stuff_1), gen2(searching_stuff_2))))

为了证明我的观点，我们可以改为searching_stuff：

async def searching_stuff_1():
    print("searching_stuff_1 begin")
    # searching
    yield 1
    await asyncio.sleep(1)
    # and searching
    yield 2
    yield 3
    print("searching_stuff_1 end")

async def searching_stuff_2():
    print("searching_stuff_2 begin")
    yield 4
    await asyncio.sleep(1)
    yield 5
    print("searching_stuff_2 end")

然后采取行动：

result = asyncio.get_event_loop().run_until_complete(gen_all())
print(result)

> searching_stuff_2 begin
> searching_stuff_1 begin
> searching_stuff_2 end
> searching_stuff_1 end
> [1, 2, 3, 4, 5]