使用 asyncio/aiohttp 进行多个 Websocket 流式传输答案

【问题标题】：multiple Websocket streaming with asyncio/aiohttp使用 asyncio/aiohttp 进行多个 Websocket 流式传输
【发布时间】：2021-06-18 09:43:40
【问题描述】：

我正在尝试使用 python 的 asyncio 和 aiohttp 订阅多个 Websocket 流。

当我运行下面的代码时，它只打印“a”，但控制台中没有其他内容作为输出。它不会抛出任何错误，而且我无法逐步调试，因为它是一个异步代码。

我想弄清楚问题是什么，如果有人可以提供帮助，我真的很感激。

import aiohttp
import asyncio

async def coro(event, item1, item2):
    print("a")
    async with aiohttp.ClientSession.ws_connect(url='url') as ws:
        event.set()
        print("b")
        await asyncio.gather(ws.send_json(item1),
                             ws.send_json(item2))
        async for msg in ws:
            print("c")
            print(msg)

async def ws_connect(item1, item2):
    event = asyncio.Event()
    task = asyncio.create_task(coro(event, item1, item2))
    await event.wait()  # wait until the event is set() to True, while waiting, block
    return task

async def main():
    item1 = {
        "method": "subscribe",
        "params": {'channel': "bar"}
    }
    item2 = {
        "method": "subscribe",
        "params": {'channel': "foo"}
    }
    ws_task = await ws_connect(item1, item2)
    await ws_task

asyncio.run(main())

【问题讨论】：

标签： python asynchronous websocket python-asyncio aiohttp

【解决方案1】：

您错误地调用了ws_connect。正确的方式：

async with aiohttp.ClientSession() as session:
    async with session.ws_connect('url') as was:
        ...

完整示例：

import aiohttp
import asyncio

async def coro(event, item1, item2):
    print("a")
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect('wss://echo.websocket.org') as ws:
            event.set()
            print("b")
            await asyncio.gather(ws.send_json(item1),
                                 ws.send_json(item2))
            async for msg in ws:
                print("c")
                print(msg)


async def ws_connect(item1, item2):
    event = asyncio.Event()
    task = asyncio.create_task(coro(event, item1, item2))
    await event.wait()  # wait until the event is set() to True, while waiting, block
    return task

async def main():
    item1 = {
        "method": "subscribe",
        "params": {'channel': "bar"}
    }
    item2 = {
        "method": "subscribe",
        "params": {'channel': "foo"}
    }
    ws_task = await ws_connect(item1, item2)
    await ws_task

asyncio.run(main())

【讨论】：