使用 asyncio/aiohttp 进行多个 Websocket 流传输 [英] multiple Websocket streaming with asyncio/aiohttp

问题描述

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

I am trying to subscribe to multiple Websocket streaming using python's asyncio and aiohttp.

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

When I run the below code, it only prints "a" but nothing else in the console as output. It does not throw any errors and I am not able to debug step by step since it's an asynchronous code.

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

I would like to figure out what the issue is, really appreciate it if anyone could help.

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())

推荐答案

您错误地调用了 ws_connect.正确方法:

You incorrectly invoke ws_connect. Right way:

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())

这篇关于使用 asyncio/aiohttp 进行多个 Websocket 流传输的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！