在 python3 中合并异步迭代器

Question

在 python3 中是否有合并异步迭代器的好方法或支持良好的库？

所需的行为与在 reactivex 中合并可观察对象的行为基本相同。

也就是说，在正常情况下，如果我合并两个异步迭代器，我希望生成的异步迭代器按时间顺序产生结果。其中一个迭代器中的错误应该会破坏合并后的迭代器。

（来源：http://reactivex.io/documentation/operators/merge.html）

这是我最好的尝试，但似乎有一个标准的解决方案：

async def drain(stream, q, sentinal=None):
    try:
        async for item in stream:
            await q.put(item)
        if sentinal:
            await q.put(sentinal)
    except BaseException as e:
        await q.put(e)


async def merge(*streams):

    q = asyncio.Queue()
    sentinal = namedtuple("QueueClosed", ["truthy"])(True)

    futures = {
        asyncio.ensure_future(drain(stream, q, sentinal)) for stream in streams
    }

    remaining = len(streams)
    while remaining > 0:
        result = await q.get()
        if result is sentinal:
            remaining -= 1
            continue
        if isinstance(result, BaseException):
            raise result
        yield result


if __name__ == "__main__":

    # Example: Should print:
    #   1
    #   2
    #   3
    #   4

    loop = asyncio.get_event_loop()

    async def gen():
        yield 1
        await asyncio.sleep(1.5)
        yield 3

    async def gen2():
        await asyncio.sleep(1)
        yield 2
        await asyncio.sleep(1)
        yield 4

    async def go():
        async for x in merge(gen(), gen2()):
            print(x)

    loop.run_until_complete(go())

Answer 1

您可以使用 aiostream.stream.merge:

from aiostream import stream

async def go():
    async for x in stream.merge(gen(), gen2()):
        print(x)

documentation and this 中有更多示例。

在 python3 中合并异步迭代器

Merging async iterables in python3

reactive-programming

python-3.x

python-asyncio