FastAPI+Redis BRPOP WebSocket消息丢失问题求助

我开发了一个简单的FastAPI应用，通过WebSocket路由结合Redis的BRPOP（阻塞列表弹出）实现消息推送，核心代码如下：

await ws.accept()
while True:
    queue,message = await aioredis_client.brpop(queue_name)
    ...# some stuff
    ws.send_json(response_dict)

环境与问题现象 #

使用环境为Python 3.8.5及最新版本的gunicorn、uvicorn、asyncio、uvloop、websockets、fastapi。尝试了gunicorn和uvicorn的多种配置组合（wsproto/websockets、uvloop/asyncio），但仍存在问题：

• 60%的时间运行正常，偶尔会出现调用send_json发送消息后浏览器完全接收不到的情况
• 重启gunicorn或切换为uvicorn均无法解决，只有向WebSocket狂发大量消息后，消息才会恢复正常推送
• 在循环中添加sleep也无效果

升级Python 3.10后的报错 #

升级至Python 3.10后出现如下报错栈：

File "/home/amalizzio/venv/lib/python3.10/site-packages/starlette/websockets.py", line 163, in send_text
    await self.send({"type": "websocket.send", "text": data})
  File "/home/amalizzio/venv/lib/python3.10/site-packages/starlette/websockets.py", line 85, in send
    await self._send(message)
  File "/home/amalizzio/venv/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 65, in sender
    await send(message)
  File "/home/amalizzio/venv/lib/python3.10/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 327, in asgi_send
    await self.send(data)  # type: ignore[arg-type]
  File "/home/amalizzio/venv/lib/python3.10/site-packages/websockets/legacy/protocol.py", line 635, in send
    await self.ensure_open()
  File "/home/amalizzio/venv/lib/python3.10/site-packages/websockets/legacy/protocol.py", line 935, in ensure_open
    raise self.connection_closed_exc()
websockets.exceptions.ConnectionClosedError: sent 1012 (service restart); no close frame received

切换为wsproto后出现类似报错栈：

await ws.send_text(j)
  File "/home/amalizzio/venv/lib/python3.10/site-packages/starlette/websockets.py", line 163, in send_text
    await self.send({"type": "websocket.send", "text": data})
  File "/home/amalizzio/venv/lib/python3.10/site-packages/starlette/websockets.py", line 85, in send
    await self._send(message)
  File "/home/amalizzio/venv/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 65, in sender
    await send(message)
  File "/home/amalizzio/venv/lib/python3.10/site-packages/uvicorn/protocols/websockets/wsproto_impl.py", line 322, in send
    output = self.conn.send(
  File "/home/amalizzio/venv/lib/python3.10/site-packages/wsproto/__init__.py", line 64, in send
    data += self.connection.send(event)
  File "/home/amalizzio/venv/lib/python3.10/site-packages/wsproto/connection.py", line 107, in send
    raise LocalProtocolError(
wsproto.utilities.LocalProtocolError: Event Message(data='...lots of json...', frame_finished=True, message_finished=True) cannot be sent in state ConnectionState.LOCAL_CLOSING.

排查与临时处理 #

• 切换为长轮询后仍存在相同问题，可在uvicorn TRACE中捕获HTTP连接关闭的情况
• 怀疑Redis的阻塞循环与Starlette存在兼容性问题：当多个标签页连接同一Web服务器（有足够多的worker）时，线程会丢失与浏览器的连接，新线程生成后，Redis弹出的消息被发送到无效连接中，通过打印弹出前后的日志可发现该问题逐渐加重
• 移除了循环，直接处理弹出的消息，但仍偶尔会丢失第一条消息，因此添加了25ms的阻塞等待，目前未再出现问题
• 同时增加了队列超时扫描备份队列的机制，确保消息可靠送达
• 切换为aioredis的async/await方式也未解决问题，计划抽空提交bug报告，特此求助

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