Python WebSocket回调队列跳过方案咨询：实时消息处理优化

这个需求太常见了——当你的处理任务耗时较长时，完全没必要被堆积的旧消息牵着走，只关心最新的实时数据才是合理的。你的思路方向是对的，但有几个细节需要调整才能稳定跑起来，我来给你拆解一下：

先分析你的原始问题 #

你原来的代码里，WebSocket的回调process是串行执行的——每收到一条消息，就会排队等前一个process执行完才会跑下一个。这就导致当处理一个消息要花10秒时，后面的消息全堆在队列里，等你处理完"one"，实时消息都已经到"twelve"了，完全跟不上节奏。

你的思路的优缺点 #

你想用一个共享变量保存最新消息，后台线程去处理的思路是对的，但有两个关键问题没解决：

1. 全局变量的线程安全问题：Python里如果在函数里给全局变量赋值，必须用global声明，否则会创建局部变量，导致后台线程拿不到更新后的消息。
2. 任务的持续触发：你只启动了一次task线程，处理完"one"就结束了，不会自动去处理后续的最新消息。

改进后的实现方案 #

我给你两种可行的实现方式，你可以根据自己的习惯选：

方案一：用事件+锁实现（灵活可控） #

这种方式用线程事件来通知后台线程有新消息，加锁保护共享变量的读写，确保线程安全：

import time
import threading

# 共享变量：保存最新消息，用锁保护避免多线程冲突
latest_msg = None
msg_lock = threading.Lock()
# 事件：用来通知后台线程有新消息需要处理
new_msg_event = threading.Event()

def process(msg):
    """WebSocket回调函数：更新最新消息并触发事件"""
    global latest_msg
    with msg_lock:
        latest_msg = msg
    # 触发事件，告诉后台线程可以处理新消息了
    new_msg_event.set()

def task():
    """后台处理线程：只处理当前最新的消息"""
    global latest_msg
    while True:
        # 等待新消息事件触发，没消息时线程会休眠，不占CPU
        new_msg_event.wait()
        # 清空事件，避免重复处理同一批消息
        new_msg_event.clear()
        
        # 安全获取当前最新消息
        with msg_lock:
            current_msg = latest_msg
            # 可选：如果不想重复处理同一个消息，可以把latest_msg置空
            # latest_msg = None
        
        if not current_msg:
            continue
        
        # 执行你的耗时任务
        print(f"\n{current_msg}", end=' ')
        for i in range(10):
            print(i, end=' ')
            time.sleep(1)
        print("处理完成")

# 启动后台处理线程（daemon=True表示主线程退出时自动终止）
threading.Thread(target=task, daemon=True).start()

# 下面是模拟WebSocket推送消息的代码，实际替换成你的ws.start_socket_abc即可
def simulate_websocket():
    for msg in ["one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten", "eleven", "twelve"]:
        print(f"\n推送消息: {msg}")
        process(msg)
        time.sleep(1)

simulate_websocket()

方案二：用固定长度队列实现（简洁省心） #

Python的queue.Queue本身是线程安全的，我们可以设置队列最大长度为1，这样新消息进来时会自动覆盖旧的：

import time
import threading
import queue

# 创建最大长度为1的队列，新消息会挤掉旧消息
msg_queue = queue.Queue(maxsize=1)

def process(msg):
    """WebSocket回调函数：把最新消息放入队列，挤掉旧消息"""
    try:
        # 尝试放入新消息，如果队列满了就抛出异常
        msg_queue.put_nowait(msg)
    except queue.Full:
        # 移除旧消息，放入新的
        msg_queue.get_nowait()
        msg_queue.put_nowait(msg)

def task():
    """后台处理线程：从队列取最新消息处理"""
    while True:
        current_msg = msg_queue.get()
        # 执行耗时任务
        print(f"\n{current_msg}", end=' ')
        for i in range(10):
            print(i, end=' ')
            time.sleep(1)
        print("处理完成")

# 启动后台线程
threading.Thread(target=task, daemon=True).start()

# 模拟WebSocket推送
def simulate_websocket():
    for msg in ["one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten", "eleven", "twelve"]:
        print(f"\n推送消息: {msg}")
        process(msg)
        time.sleep(1)

simulate_websocket()

两种方案的效果 #

不管用哪种方案，最终的输出都会是：

推送消息: one

one 0 1 2 3 4 5 6 7 8 9 处理完成
推送消息: twelve

twelve 0 1 2 3 4 5 6 7 8 9 处理完成

中间的"two"到"eleven"都会被自动跳过，完美符合你的需求。

总结 #

• 方案一的优势是灵活，你可以在事件触发时做更多自定义逻辑；
• 方案二更简洁，利用队列的特性自动处理消息覆盖，不需要手动加锁。

选哪种都可以，看你自己的代码风格和后续扩展需求。

内容的提问来源于stack exchange，提问作者thithien