正确使用 streamz 和 websocket

Question

我正在尝试找出使用streamz 处理流数据的正确方法。我的流数据是使用websocket-client 加载的，之后我这样做：

# open a stream and push updates into the stream
stream = Stream()

# establish a connection
ws = create_connection("ws://localhost:8765")

# get continuous updates
from tornado import gen
from tornado.ioloop import IOLoop

async def f():
    while True:
        await gen.sleep(0.001)
        data = ws.recv()
        stream.emit(data)
        
IOLoop.current().add_callback(f)

虽然这可行，但我发现我的流无法跟上流数据的速度（因此我在流中看到的数据比流数据落后几秒钟，这既是高容量又是高频率的数据）。我尝试将gen.sleep(0.001) 设置为较小的值（删除它会完全停止 jupyter 实验室），但问题仍然存在。

这是使用 websocket 连接 streamz 和流数据的正确方法吗？

Answer 1

我不认为 websocket-client 提供异步 API，因此它会阻塞事件循环。

您应该使用异步 websocket 客户端，例如 Tornado provides：

from tornado.websocket import websocket_connect

ws = websocket_connect("ws://localhost:8765")

async def f():
    while True:
        data = await ws.read_message()

        if data is None:
            break
        else:
            await stream.emit(data)

        # considering you're receiving data from a localhost
        # socket, it will be really fast, and the `await` 
        # statement above won't pause the while-loop for 
        # enough time for the event loop to have chance to 
        # run other things.
        # Therefore, sleep for a small time to suspend the 
        # while-loop.

        await gen.sleep(0.0001) 

如果您正在从/向远程连接接收/发送数据，那么您不需要休眠，因为远程连接的速度足以在 await 语句处暂停 while 循环。