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

Question

我正在按照说明 (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

现在我必须下载快照：

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

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

对于更新流我可以做的：

        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

Answer 1

感谢 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)

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

Process websocket stream while downloading multiple files using aiohttp

python

download

simultaneous

websocket

aiohttp