Python中如何从WebSockets队列获取最新消息并忽略未读消息

问题背景 #

需要解决两个核心需求：

• 当WebSocket队列存在未读消息时，直接获取服务器发送的最后一条消息
• 自动忽略并丢弃队列中其余的未读旧消息

现有代码会逐条接收所有消息，无法跳过旧消息直接处理最新的：

while True:
    msg = await ws_server.recv()
    await do_something_with_latest_message(msg)

期望实现的逻辑是每次仅获取当前最新消息，后续循环等待新消息到来：

while True:
    msg = await ws_server.recv_last_msg() # 下一次循环时应等待新消息到来，而非按后进先出顺序接收旧消息
    await do_something_with_latest_message(msg)

解决方案 #

由于大多数WebSocket库原生没有提供recv_last_msg这类方法，我们可以自己封装逻辑，核心思路是清空当前队列中的所有未读消息，只保留最后一条。

具体实现 #

以Python的websockets库为例，我们可以借助asyncio.wait_for实现非阻塞接收，循环清空队列：

import asyncio

async def recv_last_msg(ws_server):
    # 先获取第一条消息作为初始值
    last_msg = await ws_server.recv()
    # 循环尝试接收剩余未读消息，直到队列为空
    while True:
        try:
            # 设置极短超时，尝试非阻塞接收
            current_msg = await asyncio.wait_for(ws_server.recv(), timeout=0.01)
            last_msg = current_msg
        except asyncio.TimeoutError:
            # 超时说明没有更多未读消息，退出循环
            break
    return last_msg

# 使用示例
while True:
    latest_msg = await recv_last_msg(ws_server)
    await do_something_with_latest_message(latest_msg)

关键说明 #

• 超时时间可根据实际场景微调，0.01秒既能确保快速检查完队列中的旧消息，又不会影响正常的新消息接收
• 如果使用其他WebSocket库，需要根据其API调整非阻塞接收的实现方式，部分库可能提供has_pending_messages之类的方法用于检查队列状态
• 这个逻辑确保每次循环都只处理当前最新的消息，所有之前未读的旧消息都会被自动丢弃

内容的提问来源于stack exchange，提问作者Oliver Mohr Bonometti