使用Asyncio从Websocket输出变量的更好方法？

Question

我一直在玩Websocket和Asyncio，以获取从BitStamp流出的价格数据。连接到Websocket Feed后，我正在努力返回要在其他功能中使用的数据。我目前使用函数store_data（）作为占位符，以检查返回的价格数据。

我想出了一种相当粗略的方式来存储价格，其中涉及声明全局变量bitstamp_asks，然后在Websockets 中覆盖循环。

我觉得必须有更好的方法，因此任何帮助都将不胜感激。我的代码如下：

import asyncio
import websockets
import json

bitstamp_asks = []

async def bitstamp_connect():
    
    global bitstamp_asks

    uri = "wss://ws.bitstamp.net/"
    subscription = {
        "event": "bts:subscribe",
        "data": {
            "channel": "order_book_btcusd"
        }
    }

    async with websockets.connect(uri) as ws:
        await ws.send(json.dumps(subscription))

        subscription_response = await ws.recv()
        print(json.loads(subscription_response)['event'])

        while True:
            response = await ws.recv()
            bitstamp_data = json.loads(response)['data']
            bitstamp_asks = float(bitstamp_data['asks'][0][0])
            print(bitstamp_asks)
            await asyncio.sleep(0.0001)

async def store_data():
    while True:
        print(f'The stored value is: {bitstamp_asks}')
        await asyncio.sleep(0.1)

asyncio.get_event_loop().run_until_complete(asyncio.wait([bitstamp_connect(),store_data()]))

Answer 1

您可以使用 asyncio.queue.queue.queue 在任务之间共享值。例如：

import asyncio
import websockets
import json


q = asyncio.Queue()


async def bitstamp_connect():

    uri = "wss://ws.bitstamp.net/"
    subscription = {
        "event": "bts:subscribe",
        "data": {"channel": "order_book_btcusd"},
    }

    async with websockets.connect(uri) as ws:
        await ws.send(json.dumps(subscription))

        subscription_response = await ws.recv()
        print(json.loads(subscription_response)["event"])

        last_value = None
        while True:
            response = await ws.recv()
            bitstamp_data = json.loads(response)["data"]
            bitstamp_asks = float(bitstamp_data["asks"][0][0])

            if last_value is None or bitstamp_asks != last_value:
                q.put_nowait(bitstamp_asks)
                last_value = bitstamp_asks


async def store_data():
    while True:
        data = await q.get()
        print(f"The stored value is: {data}")
        q.task_done()


asyncio.get_event_loop().run_until_complete(
    asyncio.wait([bitstamp_connect(), store_data()])
)