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

问题描述

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

假设为简单起见，我希望镜像 2 个交易品种的订单簿：ETHBTC 和 DOGEBTC（实际上是 350+）。

首先我必须缓冲 websocket 订单更新流：

• wss://stream.binance.com:9443/stream?streams=ETHBTC@depth@100ms
• wss://stream.binance.com:9443/stream?streams=DOGEBTC@depth@100ms

现在我必须下载快照：

• https://api.binance.com/api/v3/depth?symbol=ETHBTC&limit=1000
• https://api.binance.com/api/v3/depth?symbol=DOGEBTC&limit=1000

一旦我有了快照，我就会将缓冲区（正在进行中）应用到它们，产生一个状态。

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

对于更新流，我可以这样做：

        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 的架构目标作斗争吗？

参考：

• https://gist.github.com/Hammer2900/2b5da5c08f6406ab49ddb02b0c5ae9f7

解决方法

感谢 Graingert 在 IRC Freenode #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)