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

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

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

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

def on_message(ws, message):
    data = json.loads(message)
    save_to_database(data)
    calculate_indicator(data)
    run_strategy(data)

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

def on_message(ws, message):
    message_queue.put_nowait(message)
def worker():
    while True:
        message = message_queue.get()
        try:
            process_message(message)
        finally:
            message_queue.task_done()

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

二、队列不要无限增长

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

from queue import Queue
message_queue = Queue(maxsize=10000)

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

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

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

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

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

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

```text

WebSocket

    ↓

快速接收

    ↓

Queue

    ↓

Worker

    ↓

业务处理

```

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

四、监控队列长度

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

```text

接收速率：received_msg/s

处理速率：processed_msg/s

队列长度：queue_size

```

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

## 总结

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

```text

快速接收

   ↓

异步处理

   ↓

有界队列

   ↓

监控队列长度

   ↓

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

```

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