使用 asyncio 同时监听两个 websocket

Question

我正在尝试使用 asyncio 同时收听两个 websocket。这行不通。我该怎么做？

我想获取先出现的消息，然后将其打印出来。

async with websockets.connect(sock_uri_1) as w1, \
            websockets.connect(sock_uri_2) as w2:

        msg1 = ensure_future(w1.recv())
        msg2 = ensure_future(w2.recv())
        while True:
            if msg1.done():
               print(msg1.result())
               msg1 = ensure_future(w1.recv())
            if msg2.done():
               print(msg1.result())
               msg2 = ensure_future(w2.recv())
            time.sleep(1) # Want to avoid this!

Answer 1

首先，您应该使用await asyncio.sleep(1) 而不是time.sleep(1)。通过使用asyncio.wait(when_done=FIRST_COMPLETED) 等待一条（或两条）消息到达，您的方法可以在不休眠的情况下工作：

async with websockets.connect(sock_uri_1) as w1, \
            websockets.connect(sock_uri_2) as w2:
    msg1 = ensure_future(w1.recv())
    msg2 = ensure_future(w2.recv())
    while True:
        await asyncio.wait([msg1, msg2], when_done=asyncio.FIRST_COMPLETED)
        if msg1.done():
           print(msg1.result())
           msg1 = ensure_future(w1.recv())
        if msg2.done():
           print(msg1.result())
           msg2 = ensure_future(w2.recv())

我发现管理明确的任务有些乏味，尤其是在来源数量可能不同的情况下。在这种情况下，我会使用队列将消息组合到单个通道中：

async with websockets.connect(sock_uri_1) as w1, \
            websockets.connect(sock_uri_2) as w2:
    channel = asyncio.Queue()
    async def transmit(w, source):
        while True:
            msg = await w.recv()
            await channel.put((source, msg))
    asyncio.create_task(transmit(w1, 'source1')
    asyncio.create_task(transmit(w2, 'source2')
    while True:
        source, msg = await channel.get()
        if source == 'source1':
            print(sock_uri_1, msg)
        elif source == 'source2':
            print(sock_uri_2, msg)