如何从多个 websocket 流中提取数据以在公式中使用？

Question

以下代码试图从“连接”集中获取流消息。我试图弄清楚如何从每个流中提取两个浮点值来计算商。聚合的流消息看起来是无序的和非连续的，如果我声明全局变量并尝试迭代地填充它们，我的例程就会停止。我希望能够连接到多个流并使用消息数据生成实时计算值/结果。如果可以的话请帮忙。我是 python 和 asyncio 的新手。

代码如下：

import asyncio
import websockets
import json

connections = set()
connections.add("mystream1")
connections.add("mystream2")

async def handle_socket1(url):
    async with websockets.connect(url) as websocket1:
        async for message in websocket1:
            json_msg = json.loads(message)
            if json_msg["X"] == "A":
                value1 = float(json_msg["b"])
            if json_msg["Y"] == "B":
                value2 = float(json_msg["a"])
            print(value2/value1)

async def handler1():
    await asyncio.wait([handle_socket1(url) for url in connections])

Answer 1

您可以将两个流汇集到一个队列中，并有一个单独的函数将任一值拉出队列并重新计算商：

async def handle_stream(url, identifier, queue):
    async with websockets.connect(url) as websocket1:
        async for message in websocket1:
            json_msg = json.loads(message)
            await queue.put((identifier, json_msg["value"]))

async def calculate(queue):
    value_a = value_b = None
    while True:
        identifier, value = await queue.get()
        if identifier == 'A':
            value_a = value
        else:
            value_b = value
        if value_a is not None and value_b is not None:
            print(value_a / value_b)

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        handle_stream("mystream1", "A", queue),
        handle_stream("mystream2", "B", queue),
        calculate(queue),
    )