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

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

是的，您仍然可以在异步版本中生成这些数字，这是Asynchronous Generators，您可以使用async for(PEP492)和Asynchronous Comprehensions(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只以异步方式收集协同程序的结果，所以你需要在调用asyncio.gather之前将async生成器的每个产生的结果与async def gen2分开组合，

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]