先说我上周遇到的一个真实问题。公司一个基于 FastAPI 的 IM 即时聊天后端在灰度期间陆续有用户反馈给同事发消息页面上文案已经变成“已发送”但对方手机端就是弹不出来。第一反应是客户端推送通道的问题可是查了推送厂商的后台根本没有下发记录。也就是说消息在源头上就被“吞”掉了。这条线上我前后查了快两天最后发现 FastAPI 进程本身并不背主要的锅锅在连接生命周期、进程模型和数据写入时序这些常规文档很少写透的地方。这篇文章就以 FastAPI 项目里的 IM 即时聊天接口为例把消息丢失的几条典型路径、对应的排查手段和最终落地的可靠性方案一起讲清楚。适合正在用 FastAPI 做聊天、客服系统、通知推送或群发系统的后端开发者尤其是已经能跑通 demo、但开始纠结消息可靠性的阶段。1. 一个看似简单的问题消息从发出到收到中间到底经过了几道手很多人第一次用 FastAPI 做 IM 项目时脑子里对消息链路的想象是一条直线客户端 A 把消息发到 FastAPI 接口接口找到客户端 B 的 WebSocket 连接然后把消息原封不动推过去完事。我一开始也是这么想的所以遇到“消息丢失”时第一反应是去查send_text有没有报错。但真实的生产链路上一条消息从 A 的屏幕走到 B 的屏幕至少要经过五个环节任何一个环节出问题用户体验都是“消息丢了”。客户端 A 发起发送可能是 WebSocket也可能是 HTTP 接口取决于你的架构。FastAPI 接入层接收请求路由、参数校验、中间件这里可能因为请求体截断或异常被吞而丢失。业务逻辑处理调用数据库写入、调用连接管理器推送这段是丢消息的高发区。消息路由与推送通过内存连接表或者 Redis 找到目标用户对应的连接。客户端 B 接收与展示B 可能断网、App 被系统杀死、WebSocket 连接已失效但客户端没有感知。为了直观我把每个环节常见的“丢消息姿势”整理成了表格排查时可以直接对照环节典型载体丢消息的常见姿势A 端发送WebSocket / HTTP网络闪断客户端误把“本地已写”当成“服务端已收”FastAPI 接入层路由 / Uvicorn异常被吞、请求体被反代截断、处理函数静默失败业务逻辑层async 视图 / WebSocket endpoint数据库事务回滚但接口仍返回成功、推送逻辑抛异常未捕获路由与推送内存连接表 / Redis查不到目标在线连接、广播到已失效的“幽灵连接”B 端接收WebSocket / 推送 SDK客户端断线重连期间消息无人接收、连接被系统静默杀死这里我想先给出一个结论在我排查过的 FastAPI 聊天项目里90% 的消息丢失都不是发生在 FastAPI 框架本身而是发生在“连接状态管理”和“写库与推送的顺序”上。框架不会主动吞消息吞消息的是我们自己写的连接表、异常处理和数据写入逻辑。FastAPI 从入门到实战的教程很多比如用 uv 包管理器创建虚拟环境、跑起一个 FastAPI 接口、再整合 SQLAlchemy 做用户表这些大家都能很快跑通。但 IM 场景和普通 CRUD 接口有本质区别CRUD 接口的返回值是数据本身而 IM 接口的返回值只是一个“意图”真正的数据还要经过一层长连接投递。这层投递如果不做可靠性设计消息就会在你看不见的地方消失。2. 第一现场WebSocket 连接生命周期里的幽灵连接先说一个我见过无数次的实现。很多 FastAPI 聊天项目里的连接管理器长这样from fastapi import WebSocket class ConnectionManager: def __init__(self): self.active_connections: dict[str, WebSocket] {} async def connect(self, user_id: str, ws: WebSocket): await ws.accept() self.active_connections[user_id] ws def disconnect(self, user_id: str): self.active_connections.pop(user_id, None) async def send_to_user(self, user_id: str, message: str): ws self.active_connections.get(user_id) if ws is None: return False await ws.send_text(message) return True这段代码看起来没毛病但线上跑起来就是会丢消息。问题出在哪我第一次排查时盯着send_to_user看了很久单测也过了可一到长时间运行就开始丢。2.1 连接没有及时清理广播消息落进无效通道客户端退出页面时如果前端没有主动触发 close 事件或者用户直接把 App 滑掉、电脑合盖服务端的disconnect事件可能不会立刻触发。这时候active_connections里还挂着一个已经死掉的 WebSocket 对象。后续消息推到这个连接上时底层 WebSocket 库会抛RuntimeError: Cannot call send once a close message has been sent/received。如果你在调用处只 catch 了WebSocketDisconnect这个异常会直接冒出来如果像某些项目那样写了个except Exception: pass异常倒是被吞了但这条消息也就跟着丢了。我的建议是写一个统一的safe_send包装函数发送前检查连接状态发送时捕获所有可能出现的底层异常from starlette.websockets import WebSocketState from fastapi import WebSocketDisconnect async def safe_send(ws: WebSocket, payload: str) - bool: if ws.application_state ! WebSocketState.CONNECTED: return False try: await ws.send_text(payload) return True except (RuntimeError, WebSocketDisconnect): return False这里有个细节值得注意WebSocketState.CONNECTED只是告诉你连接对象曾经成功握手过并不能保证连接此刻一定还活着。所以底层 send 时的异常捕获是必须的状态检查只是用来减少无效调用的次数。2.2 多设备登录时连接被后登录的设备顶掉另一个隐藏得很深的坑是连接表的 key 设计。如果直接用user_id作为字典的 key同一个用户从 PC 和手机同时登录时后建立的连接会覆盖先建立的连接。比如用户 B 先在 PC 端登录又在手机端登录服务端连接表里 B 对应的 WebSocket 就变成了手机端的连接。这时候用户 A 给 B 发消息服务端把消息推到了 B 的手机端但 B 的手机端可能被系统折叠了通知或者 B 根本不看手机只盯着 PC 屏幕。B 的感知就是“我没有收到消息”消息实际上进了另一个设备。解决思路是把连接 key 细化到设备维度self.active_connections[f{user_id}:{device_id}:{conn_id}] ws同时维护一个user_id - set[conn_key]的映射广播时遍历该用户的所有连接。是否需要多端同时收到消息、还是只推最新连接这是产品决策但后端至少不应该用覆盖 key 的方式把其他设备静默踢掉。2.3 只转发不落库断线期间消息彻底蒸发这是我见过的最离谱、但也最常见的丢消息方式。有些 FastAPI 聊天 demo 为了演示方便WebSocket endpoint 里收到消息后直接调用send_to_user转发给目标用户完全不做数据库持久化。这种“转发型 IM”在联调时一切正常因为两个客户端都连着同一个服务端。一旦某个用户中途断网或者 WebSocket 连接刚好在发送瞬间断开消息既没有落库也没有离线缓存直接就在网络里消失了。用户重连之后什么都拉不回来。生产环境里消息至少要做到先落库再推送。落库不只是为了历史记录更是为了给“连接不可达”情况下的离线补拉提供一个兜底。只转发不落库的架构无论你怎么优化连接管理消息丢失都只是时间问题。3. 进程模型盲区多 Worker 下内存连接表各说各话如果说幽灵连接是“单个进程内部的坑”那么多 Worker 部署就是“跨进程的坑”。这个坑特别隐蔽因为开发环境和单机测试时根本不会暴露。3.1 为什么单机多 Worker 也会丢消息假设你部署 FastAPI 服务时使用了以下命令uvicorn main:app --workers 4Uvicorn 会启动 4 个独立进程每个进程都有自己的事件循环和独立的内存空间。WebSocket 长连接在被操作系统负载均衡到某个 worker 后就会一直待在那个进程里。问题来了你的ConnectionManager是进程内存变量。用户 A 的连接可能落在了 worker 1用户 B 的连接落在了 worker 2。当 A 发消息时请求如果被分配到 worker 1worker 1 在自己的内存连接表里根本找不到 B 的连接消息就被判定为“目标不在线”。更麻烦的是这种情况在 Redis 或数据库里查在线状态时B 明明是在线的。因为在线状态和实际连接放在两个不同的地方状态判断和消息推送之间出现了断层。在我处理过的一个项目里线上部署 4 个 worker消息丢失率在高峰期能达到 2% 左右。平时流量低worker 之间的连接分布不均匀偶尔丢一条也没人注意流量一上来某个 worker 上的连接特别多跨 worker 的消息全部丢失用户立刻炸锅。3.2 用 Redis 做跨进程路由的落地方法解决多 worker 下连接表不一致核心思路是把“在线连接在哪台机器/哪个 worker”这个信息从本地内存搬到所有 worker 都能访问的地方。最轻量的做法是用 Redis Pub/Sub 做广播。基本思路是这样每个 worker 启动后订阅一个全局的消息推送频道。业务侧需要给某个用户推消息时不直接查本地连接表而是把消息发布到 Redis 频道。所有 worker 都会收到这个推送通知然后检查目标连接是否在自己的本地连接表里。如果在就走safe_send如果不在说明目标用户不在这个 worker 上直接忽略等离线补拉逻辑兜底。示意代码如下import redis.asyncio as aioredis class RedisRouter: def __init__(self, redis_url: str): self.redis aioredis.from_url(redis_url) async def publish(self, channel: str, message: str): await self.redis.publish(channel, message) async def subscribe(self, channel: str): pubsub self.redis.pubsub() await pubsub.subscribe(channel) return pubsub每个 worker 里再跑一个后台任务消费频道消息async def push_listener(manager: ConnectionManager, redis_url: str): router RedisRouter(redis_url) pubsub await router.subscribe(im:push) async for message in pubsub.listen(): if message[type] ! message: continue # message 里包含 user_id 和 payload # 按 user_id 在本地连接表里查找并发送这里必须强调一点Redis Pub/Sub 是“即发即弃”的如果某个 worker 正在重启或者订阅连接断开了这期间的消息不会补发给它。所以 Pub/Sub 只能解决“找到连接”的问题不能解决“消息不丢”的问题。真正的兜底仍然要靠数据库落库和客户端上线补拉。我实际测试过一个 3 worker 的 FastAPI 聊天服务改造前并发 200 连接时消息丢失明显改造后同样的压测场景连续跑 30 分钟一条都没丢。这说明跨进程路由确实是多 worker 场景下的关键。另外提一个开发期的小坑很多人喜欢用uvicorn main:app --reload启动热更新本地跑得飞起因为 reload 模式下 Uvicorn 默认禁用多 worker部署时改成--workers 4行为就完全不一样。如果你在本地没有模拟过多 worker 环境很多连接路由问题根本发现不了。4. 写库与推送之间的空窗接口返回成功不等于对方已送达下面说一个很多人都踩过、但未必意识到的逻辑问题IM 接口的“成功返回”到底代表什么4.1 先推后存与先存后推的取舍常见的 FastAPI 消息发送接口一般长这样app.post(/messages) async def create_message(payload: MessageCreate): msg await save_to_db(payload) await connection_manager.send_to_user(payload.receiver_id, msg.json()) return {code: 0, msg_id: msg.id}这段代码有两个隐患。第一send_to_user返回False时接口照样返回{code: 0}。发送方看到的是“已发送”接收方却永远收不到。你可能会说在用户界面上显示“已发送”不代表对方已读这是常识。但对很多业务系统来说“已发送”意味着服务端已经把消息推给目标连接了现在推送实际失败了这会被误判。第二消息先落库、后推送但如果推送抛异常异常被某个上层中间件吞掉接口依然返回成功。数据库里有这条消息推送日志里却没有成功记录消息就躺在了“已入库但未投递”的中间态。如果反过来先推后存问题更严重推送成功但写库失败接收方看到了消息发送方记录里却找不到重新登录后聊天记录丢失这也是另一种“消息丢失”。所以正确的做法不是简单调整顺序而是给消息定义明确的状态并让接口返回值只代表“服务端已接收并落库”不代表“对方已经收到”。4.2 只把推送失败记进日志迟早会被日志淹没我之前在一个项目里看到过这样的代码try: await connection_manager.send_to_user(receiver_id, message) except Exception: logger.error(send failed, exc_infoTrue)日志打了消息丢了用户来投诉你从几万条日志里翻出这条报错除了证明“我们确实出过错”之外没有任何帮助。因为消息已经不在待重试队列里也没有离线标记错过的就永远错过了。要改变这个局面消息表必须有一个状态字段比如status。我常用的状态设计是pending服务端已落库尚未尝试推送。pushed服务端已通过连接推送但尚未收到接收方确认。delivered接收方客户端已确认收到。read接收方已读。推送失败时不只写日志还要把状态置回pending或者明确标记为“待重试”。这样后台可以定时扫描把没有送达的消息重新投递。补一条条件更新 SQL 供参考UPDATE messages SET status delivered, delivered_at NOW() WHERE msg_id %s AND receiver_id %s AND status pending这个 WHERE 条件保证了状态流转是幂等的即使同一个回执被客户端重复提交也不会把消息状态搞乱。4.3 离线消息归档与增量拉取有了状态标记离线补拉就顺理成章了。接收方客户端重新连接 WebSocket 之后第一步不是干等新消息而是调用一个“拉取离线消息”的接口GET /messages?user_id1002after_seq10240接口返回该用户在断线期间收到的、status不为delivered的消息客户端渲染后再逐条上报回执。这个设计把“推送失败”变成了“暂存待取”消息不会因为在推送环节出错就消失。离线消息不能无限期保留一般我按业务需求设置保留窗口比如 7 天或 30 天超过窗口的消息要么归档到冷存储要么直接清理。这个窗口要大于业务上“用户最长离线时间”否则用户隔了很久重新登录却发现更早的消息不在了体验同样不好。5. 反向代理和 Uvicorn 的默认超时连接被悄悄掐断的隐蔽入口消息丢失还有一种极其隐蔽的来源就是连接本身没有异常但被中间设备“静默断开”了。5.1 Nginx 默认 60 秒空闲断连很多 FastAPI 服务前面会挂一层 Nginx 做反向代理。如果你用默认配置代理 WebSocket那么一个非常容易被忽略的参数是proxy_read_timeout默认值是 60 秒。这意味着什么如果客户端与 WebSocket 服务端之间连续 60 秒没有任何数据流动Nginx 会主动关闭这条连接。WebSocket 本身是个长连接聊天场景里用户可能连续几分钟不说话连接刚好在后台被掐断。客户端如果没有及时发现服务端也不知道连接已失效后续的消息推送就全部落入虚空。正确的 Nginx 配置至少应该包含以下几项location /ws/ { proxy_pass http://fastapi_backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; proxy_read_timeout 3600s; proxy_send_timeout 3600s; }注意Connection: upgrade头是针对 WebSocket 协议升级的HTTP/1.0 默认不支持必须显式指定proxy_http_version 1.1。我见过有人只加了 Upgrade 头、忘了把 HTTP 版本改成 1.1结果 WebSocket 一直握手失败但这个错误通常会在联调期暴露不会拖到线上。5.2 心跳机制不是可选项是必选项有了 Nginx 超时配置还不够。即使你把超时时间拉长到 3600 秒一条连接持续空闲几小时中间任何一层设备都可能因为资源回收策略把它断开。服务端和客户端之间必须有心跳。Uvicorn 自身支持 WebSocket 层的 ping/pong 参数例如uvicorn main:app --ws ping-interval 20 --ws ping-timeout 20这样可以保证底层连接定期有数据包流动减少被中间设备判定为“空闲连接”的概率。但底层 ping 只能维持连接存活业务层心跳还有另一个作用让服务端能主动清理死连接。我在实现时通常让客户端每 30 秒发送一个应用层心跳消息服务端收到后更新last_seen时间戳后台定时任务每隔一段时间扫描一次把超过 90 秒没有心跳的连接从连接表里移除。这样即使客户端异常退出、服务端没收到 disconnect 事件幽灵连接也会在一个心跳周期内被清理掉。这个清理动作的价值在于它把“连接不可用”的发现时间从“下一次推送报错”提前到了“心跳超时”。推送报错时已经有一条消息被牺牲了而心跳超时清理是在消息到来之前就把问题解决了。6. 一条消息的完整投递契约ACK、去重与重试设计前面讲的都是排查和局部修复下面说整体方案。要让 IM 消息在 FastAPI 项目里不丢单纯靠“改一个连接表”是不够的需要把可靠性设计到消息的整个生命周期里。我通常称之为“投递契约”。6.1 消息 ID 与服务端 ACK客户端发送消息时不应该等服务器返回后再生成消息 ID而应该在本地生成一个client_msg_id客户端消息 ID。这个 ID 的作用是让重试和去重有据可依。服务端收到消息后先根据client_msg_id做幂等检查。如果 Redis 或数据库里已经有这个 ID直接返回已存在的server_msg_id不再重复落库。然后落库、推送最后返回 ACK 给发送方。ACK 里至少要包含三样东西client_msg_id告诉客户端“你发的那条消息我收到了”。server_msg_id服务端生成的消息主键。server_time服务端接收时间用于客户端做消息排序。发送方的 UI 只有拿到这个 ACK 才能把状态从“发送中”改成“已发送”。如果客户端在超时时间内没收到 ACK就自动重试或提示用户手动重发。这是防止“客户端误报成功”的关键。6.2 接收方回执与状态流转服务端推送消息给接收方后消息状态是pushed还不是delivered。真正的“送达”要以接收方客户端上报为准。接收方收到消息并成功渲染后客户端应向服务端发一条回执消息内容至少包括server_msg_id。服务端收到回执后把消息状态更新为delivered。如果产品还要做“已读”那就等用户真正打开聊天窗口后再上报已读状态更新为read。整个状态流转可以用一张表说明阶段发送端表现服务端状态接收端表现客户端发送显示“发送中”pending-服务端落库并返回 ACK显示“已发送”pending/pushed-服务端推送接收方收到显示“已发送”pushed显示新消息接收方上报回执显示“已送达”delivered-接收方已读上报显示“已读”read-这里有一个容易被忽略的细节回执和消息本身一样也可能丢失。所以接收方在上报delivered或read时同样可以做重试。但重试必须配合幂等更新也就是前面那条带 WHERE 条件的 SQL避免把状态从read回退到delivered。6.3 重试与幂等重试机制里最怕的是“火上浇油”。如果没有幂等控制客户端断网重试时同一条消息可能被服务端落了两次库接收方会看到两条一模一样的消息。用户不会认为这是“消息重复”而会觉得你的系统有问题。所以client_msg_id必须作为唯一键约束落在数据库里。第二次收到同样的client_msg_id时直接返回第一次生成的server_msg_id不再生成新消息。服务端推送失败后的重试也应该有上限和退避策略。比如每 5 秒重试一次最多重试 6 次超过次数后不再盲目重推而是把消息留在pending状态等接收方上线后主动拉取。这种“推送重试 离线补拉”双轨并行的方式能覆盖绝大多数复杂网络场景。7. 验证与回归用日志链路和故障注入证明丢消息问题被修复很多人在修改完代码后跑一遍正常流程发现“能收发消息”就认为问题解决了。但丢消息这类问题恰恰要在异常场景下才会暴露。我后来总结了一套验证方法推荐给所有搭 IM 后端的人。7.1 日志链路每条消息都要有完整轨迹给所有消息相关日志加上统一的msg_id字段。从接入层收到消息开始到落库、推送、ACK、回执每一步都打印一条结构化日志msg_id20250120001 actionreceived from1001 to1002 msg_id20250120001 actionpersisted statuspending msg_id20250120001 actionpush_start msg_id20250120001 actionpush_success排查问题时只用一条命令就能看到某条消息的完整生命周期grep msg_id20250120001 app.log如果发现actionpush_success之后没有对应的回执日志说明接收方可能没收到或者收到了但回执丢失。整个过程一目了然。7.2 故障注入测试清单我每改一次消息可靠性相关的代码都会跑一遍下面这张测试清单模拟真实场景里的各种故障场景操作预期结果接收方断网直接关闭接收端 WebSocket然后发送消息消息落库为 pending接收方重连后可补拉发送方超时服务端暂停返回 ACK客户端重试服务端按 client_msg_id 幂等去重多 worker 路由3 个 worker 启动随机建立 200 个连接所有在线用户消息均能收到无跨进程丢失反向代理空闲断开心跳暂停 90 秒服务端清理死连接后续推送不再进入该通道重复回执接收方连续上报两次 delivered消息状态保持 delivered不会回退或报错清单里每一项都可以通过自动化脚本跑。我实际遇到过一种尴尬情况改完代码后所有手工测试都通过但线上还是会丢消息后来发现是压测工具只创建了一个 worker 的进程根本没有触发跨进程问题。所以故障注入场景里一定要显式让多 worker 同时在线。7.3 自动化回归测试脚本的思路用 pytest pytest-asyncio 可以写一套针对消息可靠性的回归测试。核心是用两个 WebSocket 客户端分别模拟发送方和接收方然后强制断开重连验证消息能否补拉回来。async def test_message_delivery_after_reconnect(): sender_ws await connect_ws(/ws/1001) receiver_ws await connect_ws(/ws/1002) # 正常发送场景 await sender_ws.send_text(json.dumps({ client_msg_id: c1, to: 1002, content: hello })) msg await receiver_ws.receive_text() assert json.loads(msg)[client_msg_id] c1 # 接收方断线发送方再发一条 await receiver_ws.close() await sender_ws.send_text(json.dumps({ client_msg_id: c2, to: 1002, content: hello2 })) ack await sender_ws.receive_text() assert json.loads(ack)[status] sent # 接收方重连后拉取离线消息 receiver_ws2 await connect_ws(/ws/1002) await receiver_ws2.send_text(json.dumps({ action: pull_offline, last_seq: 0 })) data await receiver_ws2.receive_text() assert c2 in data这套脚本跑通了说明可靠性改动没有破坏基本功能也在反复构建中替我挡掉过很多回归问题。最后再分享一个排查时的额外收获IM 消息丢失的问题表面上是“某一行代码没有处理好”实际上往往是连接状态、进程模型、数据时序三个层面叠加的结果。修 bug 时最好把这三层全部过一遍而不是只盯着当前报错的位置。我那次排查到最后真正的问题其实是三个小问题同时发生连接表 key 被覆盖、多 worker 内存隔离、推送失败没有重试标记。单独看每一个都算不上严重三个叠在一起就是线上事故。这也是我为什么建议把“消息可靠性”作为一个整体架构去设计而不是零散地打补丁。