WebSocket实时数据接收后如何避免消息堆积？

做实时行情接入时，我遇到过一个比较典型的问题：WebSocket 一直保持连接，数据也在正常接收，但运行一段时间后，消息延迟越来越明显，内存占用也不断上涨。

后来发现，问题是数据接收速度超过了业务处理速度。比如每秒收到 1000 条消息，但程序只能处理 700 条，那么每秒就会多出 300 条，时间一长自然会产生积压。

一、不要在WebSocket回调里做重计算

我比较推荐把“接收”和“处理”拆开。不建议直接在 `on_message()` 中进行数据库写入、复杂计算或者策略判断：

1. <div align="left">def on_message(ws, message):</div><div align="left">    data = json.loads(message)</div><div align="left">    save_to_database(data)</div><div align="left">    calculate_indicator(data)</div><div align="left">    run_strategy(data)</div>

复制代码

更合理的方式是先把消息放入队列，再由 Worker 负责处理：

1. <div align="left">def on_message(ws, message):</div><div align="left">    message_queue.put_nowait(message)</div><div align="left">def worker():</div><div align="left">    while True:</div><div align="left">        message = message_queue.get()</div><div align="left">        try:</div><div align="left">            process_message(message)</div><div align="left">        finally:</div><div align="left">            message_queue.task_done()</div>

复制代码

这样即使数据库或计算逻辑偶尔变慢，也不会直接阻塞 WebSocket 的消息接收。

二、队列不要无限增长

我通常会给队列设置一个上限：

1. from queue import Queue
   
2. message_queue = Queue(maxsize=10000)
   

复制代码

无限队列看起来可以暂时缓解问题，但如果下游长期处理不过来，消息会持续占用内存，最终可能拖垮进程。

队列满了以后怎么处理，需要根据业务决定。如果是实时行情看板，通常可以只保留最新行情；如果是 Tick 数据存档或回测，则不能随意丢数据，而应该通过增加 Worker、批量写入或消息队列等方式提高消费能力。

三、一个通用的WebSocket处理示例

下面使用通用 WebSocket 地址作为 placeholder，不绑定具体行情服务：

1. <div align="left">import json</div><div align="left">import queue</div><div align="left">import threading</div><div align="left">import websocket</div><div align="left">TOKEN = "YOUR_TOKEN"</div><div align="left">WS_URL = "wss://your-websocket-server.example.com/ws"</div><div align="left">message_queue = queue.Queue(maxsize=10000)</div><div align="left">def process_message(message):</div><div align="left">    data = json.loads(message)</div><div align="left">    print("processing:", data)</div><div align="left">def worker():</div><div align="left">    while True:</div><div align="left">        message = message_queue.get()</div><div align="left">        try:</div><div align="left">            process_message(message)</div><div align="left">        finally:</div><div align="left">            message_queue.task_done()</div><div align="left">def on_open(ws):</div><div align="left">    subscribe_message = {</div><div align="left">        "action": "subscribe",</div><div align="left">        "symbol": "YOUR_SYMBOL",</div><div align="left">        "token": TOKEN</div><div align="left">    }</div><div align="left">    ws.send(json.dumps(subscribe_message))</div><div align="left">def on_message(ws, message):</div><div align="left">    try:</div><div align="left">        message_queue.put_nowait(message)</div><div align="left">    except queue.Full:</div><div align="left">        print("Message queue is full")</div><div align="left">def on_error(ws, error):</div><div align="left">    print("WebSocket error:", error)</div><div align="left">def on_close(ws, close_status_code, close_msg):</div><div align="left">    print("WebSocket closed:", close_status_code, close_msg)</div><div align="left">for _ in range(4):</div><div align="left">    threading.Thread(target=worker, daemon=True).start()</div><div align="left">ws = websocket.WebSocketApp(</div><div align="left">    WS_URL,</div><div align="left">    on_open=on_open,</div><div align="left">    on_message=on_message,</div><div align="left">    on_error=on_error,</div><div align="left">    on_close=on_close</div><div align="left">)</div><div align="left">ws.run_forever()</div>

复制代码

这个示例的核心不是 Worker 一定要设置为 4 个，而是把处理链路拆成：

```text

WebSocket

    ↓

快速接收

    ↓

Queue

    ↓

Worker

    ↓

业务处理

```

WebSocket 回调只负责快速接收消息，耗时操作交给后面的处理层。

四、监控队列长度

实际部署以后，我建议至少关注三个指标：

```text

接收速率：received_msg/s

处理速率：processed_msg/s

队列长度：queue_size

```

如果接收速度长期高于处理速度，单纯扩大 Queue 只能延缓问题，不能真正解决消息堆积。这时候应该继续排查数据库、计算逻辑、网络请求等环节，找到真正的性能瓶颈。

## 总结

WebSocket 消息堆积，本质上是**生产速度和消费速度不匹配**。我的处理思路通常是：

```text

快速接收

   ↓

异步处理

   ↓

有界队列

   ↓

监控队列长度

   ↓

根据业务决定丢弃、合并或完整处理

```

对于实时行情和 Tick 数据来说，关键不是简单把队列做得更大，而是让 WebSocket 接收层足够轻，并确保后面的消费能力能够跟上数据产生速度。