使用aiohttp下载多个文件时处理websocket流

Process websocket stream while downloading multiple files using aiohttp

我正在按照说明 (here) 在我的本地计算机上镜像 Binance Exchange 上的多个订单簿。

假设为简单起见,我希望镜像 2 个符号的订单簿:ETHBTC 和 DOGEBTC(实际上是 350+)。

首先我必须缓冲 websocket 顺序更新流:

现在我必须下载快照:

一旦我有了快照,我就将缓冲区(正在进行中)应用于它们,产生一个状态。

之后,所有订单更新都可以简单地应用于状态。

对于更新流我可以做的:

        async with aiohttp.ClientSession() as session:
            async with session.ws_connect(URL) as wsock:
                async for msg in wsock:
                    if msg.type != aiohttp.WSMsgType.TEXT:
                        J = json.loads(msg.data)
                        symbol = J['data']['s']

                        process_update(symbol, J)

但是,一旦第一个更新进入,我如何开始下载快照,并使用完成处理程序来处理它,而不中断流?

如果我跟踪 300 个符号,则表示同时进行了 300 次下载。

我找到了有关异步下载多个文件的资源,但我看不出如何将其与处理流的要求相结合。

我总是可以在单独的线程中进行下载,但这不是违背了 aiohttp 的架构目标吗?

参考文献:

感谢 IRC Freenode 上的 graingert 回答 #python

import anyio.to_thread


async def foo():
    async def download(symbol):
        async with session.get(f"{url}/{symbol}") as resp:
            await do_something(resp)

    async with aiohttp.ClientSession() as session, session.ws_connect(
        URL
    ) as wsock, anyio.create_task_group() as tg:
        async for msg in wsock:
            if msg.type != aiohttp.WSMsgType.TEXT:
                J = json.loads(msg.data)
                symbol = J["data"]["s"]

                tg.start_soon(download, symbol)
                await anyio.to_thread.run(process_update, symbol, J)