项目收尾那天客服主管跑过来问我“你们说的 AI 客服助手用户聊到一半断网再回来之前的内容能续上吗”我当时一愣因为演示环境一直走的是内网断线续聊这块还真没单独压过。这个问题直接把 Spring AI 客服项目从“把功能跑通”推向了“把消息链路做可靠”后面两周全部铺在事件总线、Redis Pub/Sub 和断线补拉这三件事上。这篇就是这次企业级收官的完整复盘包括为什么走这套方案、代码怎么组织、上线前踩了哪些坑以及补拉压测的真实数字。1. 客服会话消息为什么会“走着走着就丢了”1.1 最初架构WebSocket 直连大模型调用这个客服项目最初的链路很简单用户在前端页面发起会话消息通过 WebSocket 推到后端后端同步调用大模型拿到回复后再通过同一个 WebSocket 通道推回给用户。Spring AI 在这里主要负责把大模型接入封装成模板化调用对业务代码来说就是一个 ChatClient传 prompt 进去、拿 content 出来。这套架构在 Demo 阶段完全够用因为单实例部署、内网环境稳定、消息量也小。但真正往企业级方向收敛时问题立刻暴露了WebSocket 连接是长连接而大模型调用是典型的慢操作动不动三五秒高峰期甚至能到十秒以上。同步调用的时候只要客户端在等待期间断一次网、切一次后台、或者代理超时这次回复就再也回不到用户屏幕上。用户看到的结果就是“问完问题没反应”只好再问一遍然后系统又调一遍大模型既浪费钱又伤体验。更麻烦的是一旦把服务扩展成多实例部署同步链路会变得根本说不通。用户连的是节点 AAI 回复如果由节点 B 的异步任务产生节点 B 怎么知道该往哪里推所以最初的实现里很多回复干脆就是“查询最新一条”这种绕法本质上是在用数据库轮询解决实时推送问题。1.2 线上差点翻车的一个真实场景让我决定重构的直接触发点是一次内部模拟。我们开着两个浏览器窗口扮演坐席和客户客户这边发完消息后立刻断网 30 秒等网络恢复再刷新页面。结果会话记录里只有用户自己发的那句话AI 回复彻底没了坐席端也看不到上下文整个对话变成单方面输出。排查代码后根因非常直白AI 回复是同步写回 WebSocket 的断线期间没有任何补偿手段消息发出去了客户端没收到服务端也没有记录“这条消息投递成功还是失败”。在长连接场景里没有确认机制就等于没有可靠性这是架构层面的缺口不是加几个重试就能补的。1.3 改造的三个设计目标针对这个问题我们给收尾阶段定了三个明确目标后面所有设计都为这三个目标服务。第一不丢。会话内所有消息必须留痕至少保留 24 小时哪怕客户端全程离线重连后也能补齐。第二可续。断线重连后客户端要能拉到断线期间错过的消息并且对话上下文要连续不能出现“你刚问的问题我失忆了”的情况。第三可扩。多实例部署时消息能准确投递给持有该会话连接的节点新增实例不需要改业务代码只靠基础设施能力解决。这三个目标一摆出来方案其实就被锁死了应用内必须有一层事件总线把消息生产和消费解耦跨实例必须用 Redis Pub/Sub 做广播客户端侧必须引入游标机制做断线补拉。下面的内容就按这三条线展开。2. 应用内事件总线把“谁处理、何时处理”从调用方手里拿走2.1 事件模型怎么划分三类事件覆盖全链路事件总线要管的是“一条消息从用户发出到最终展示中间每一步状态变化”。我们把所有业务流转抽象成三类事件用户消息接收事件、AI 回复就绪事件、会话状态变更事件。用户消息接收事件在 WebSocket 收到消息时发布携带会话 ID、用户 ID、消息文本、客户端生成的消息 ID。AI 回复就绪事件在大模型调用返回后发布携带会话 ID、消息 ID、回复内容、token 消耗等元数据。会话状态变更事件则用来通知所有相关方“会话进入了什么阶段”比如转人工、排队中、会话关闭。事件定义直接用 Java record配合 Jackson 做序列化。这里有一个细节建议事件里一定要带统一的消息 ID 和时间戳后续做幂等和补拉游标都依赖这两个字段。我们最初只传了业务字段结果联调补拉时发现无法判断哪条消息是重复的又回头加字段属于完全可以避免的返工。2.2 异步分发与背压控制为什么必须自研而不是用注解最开始我用 Spring 的 EventListener 做事件监听但很快发现同步分发有两个硬伤监听器方法里如果调用大模型整个 WebSocket 线程会被阻塞而且监听器之间没有依赖控制处理 AiReplyReadyEvent 时如果 UserMessageReceivedEvent 还没落库状态会错乱。后来换成了自研的轻量事件总线核心就三部分一个线程池、一张“事件类型到处理器列表”的路由表、一个发布方法。发布方法把事件交给线程池异步分发处理器各自处理自己的事。线程池用 ThreadPoolTaskExecutor核心线程数按实例 CPU 核数乘 2 配置队列容量给到 2000拒绝策略用的是 CallerRunsPolicy。调用方代码大概长这样Component public class ChatEventBus { private final ThreadPoolTaskExecutor executor; private final MapClass?, ListChatEventHandler handlers new ConcurrentHashMap(); private final ChatEventStore eventStore; public ChatEventBus() { this.executor new ThreadPoolTaskExecutor(); this.executor.setCorePoolSize(8); this.executor.setMaxPoolSize(16); this.executor.setQueueCapacity(2000); this.executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); this.executor.setThreadNamePrefix(chat-event-); this.executor.initialize(); } public void publish(ChatEvent event) { eventStore.append(event); // 先落库再分发 ListChatEventHandler list handlers.get(event.getClass()); if (list null) return; for (ChatEventHandler handler : list) { executor.execute(() - handler.handle(event)); } } }选 CallerRunsPolicy 是故意的线程池满的时候不丢弃任务而是让发布线程自己处理代价是发布方变慢但消息不丢。一开始我用的 AbortPolicy结果压测时线程池一满就抛异常好几条 AI 回复直接没了这种教训一次就够了。2.3 事件持久化内存队列清空就等于丢消息事件总线解决了解耦和异步但内存队列本身是易失的进程一重启队列里还有几百条事件没处理这些消息就永远消失了。所以设计里有一条铁律重要事件必须先持久化再进入分发流程。持久化我们选的是 Redis Stream按事件类型建流每条事件写入后分配全局自增 ID。这样就算消费者进程挂了重启后还能从 Stream 的消费游标继续读。这里要区分两个角色Redis Pub/Sub 负责把事件推给所有在线实例提供实时性Redis Stream 负责把事件存下来提供可靠性和补拉数据源。两者不是替代关系而是配合关系。很多文章喜欢把 Pub/Sub 和持久化队列对立起来实战中完全不是非黑即白。Pub/Sub 的优点是实时、低延迟、广播效率高缺点是不持久化Stream 的优点是持久化、可追溯缺点是消费逻辑更重、管理和运维成本更高。客服消息这种场景在线时用 Pub/Sub 推离线后靠 Stream 补正好各取所长。3. Redis Pub/Sub 跨节点广播多实例集群的会话路由3.1 集群模式下消息怎么找到正确的连接改造完事件总线后单实例内部的实时性已经没问题但企业级部署必然是多个实例共享同一个 Redis。用户在节点 A 建立 WebSocketAI 回复可能由节点 B 的事件总线处理器生成节点 B 必须想办法把消息投给节点 A否则回复就烂在 B 的内存里了。最简单的方案是每个实例启动时向 Redis 注册“本机持有哪几个会话连接”然后根据会话 ID 精准推送。但实践下来这个方案维护成本不低连接断开会话关闭要清理注册表实例宕机的僵尸会话要定期扫描集群扩容缩容时注册表要重新分配。这相当于自己做一遍服务注册发现复杂度远超收益。Redis Pub/Sub 的广播模式恰好绕开了这个问题回复事件发送到频道后所有实例都会收到一份拷贝每个实例自己判断“这个会话的连接在不在本机”在就推给当前连接不在就丢弃。业务代码里只需要一个会话管理器维护本机 WebSocket 和会话 ID 的映射其它实例的清理工作完全不用关心。判定逻辑简化后就是这样public void onAiReply(AiReplyReadyEvent event) { WebSocketSession session sessionRegistry.get(event.getSessionId()); if (session ! null session.isOpen()) { session.sendMessage(new TextMessage(event.getPayload())); } }3.2 为什么不用数据库轮询也不用 Kafka方案评审时有人问能不能用数据库轮询替代 Redis Pub/Sub我的回答是能跑但非常不优雅。轮询要保证延迟低就得缩短间隔比如每 200ms 查一次四个实例同时查一张会话消息表Redis 本身的及时性和轻量优势全没了还白白增加数据库压力。至于 Kafka那完全是另一层考量。客服团队这个项目的集群规模就四个实例消息量一天几十万条引入 Kafka 意味着要部署 Broker、管理 Topic、处理分区和消费组运维复杂度直接上一个台阶。Redis 本来就在链路里Pub/Sub 零额外组件几百行代码就解决了先把业务价值交付出去更重要。3.3 Channel 命名与消息协议设计Pub/Sub 的 Channel 对业务不可见但命名规范直接影响排障效率。我们按事件类型划分了三个频道chat:event:user、chat:event:ai、chat:event:session。这样监听时可以按频道分别配置处理逻辑排查问题时也可以单独对一个频道做订阅测试。消息体协议统一用 JSON 字符串外层是统一信封内层是业务 payload。信封里必须包含 eventId、eventType、sessionId、timestamppayload 里至少包含 seq 和 messageId。seq 是会话内自增序号补拉游标依赖它messageId 是全局唯一消息 ID幂等去重依赖它。发布代码很简单redisTemplate.convertAndSend(chat:event:ai, eventJson);监听端配置要稍加注意Spring Data Redis 要用 RedisMessageListenerContainer指定连接工厂、监听适配器和 ChannelTopic。3.4 序列化与监听容器恢复两个最容易踩的坑第一个坑是 Redis 序列化不一致。默认的 JdkSerializationRedisSerializer 会把 Java 对象序列化成二进制Redis 里看到的是一堆\xAC\xED开头乱码而且跨语言排查问题非常痛苦。我们统一改成 Jackson JSON 序列化消息 DTO 全部用 record 或者带无参构造的普通类序列化器明确配置成 GenericJackson2JsonRedisSerializer。第二个坑是监听容器在 Redis 连接抖动后可能“假死”。连接被 Redis 服务端断开后容器本身有重连机制但如果监听器处理时异常没被捕获整个 listener 会停止消费后续消息。必须显式设置 errorHandler把所有异常打出来并且处理逻辑内部再做一层 try/catchcontainer.setConnectionFactory(redisConnectionFactory); container.setErrorHandler(e - log.error(Redis listener error, e));第三个坑是语义层面的Pub/Sub 是即发即弃模式。Redis 实例重启、网络分区、订阅方不在线这些场景下消息都会丢而且 Redis 不会给你任何“消息没送达”的回执。所以必须认清一个事实Pub/Sub 承担的是“在线加速”不是“可靠投递”可靠性要靠第四节的补拉机制兜底。4. 断线补拉给“实时推送”装一个可靠兜底4.1 游标设计与 Redis 留存结构断线补拉的本质是让客户端在重连后能告诉服务端“我最后看到的是哪条消息”服务端把比这个位置更新的消息全部返回。所以每个会话必须有一个严格递增、可比较的序号这就是 seq。为了拿到自增不重复的 seq我们用 Redis 的 INCR 命令每个会话一个计数器 key。消息无论是用户发的还是 AI 回的进留存结构前都先取 seq。存储结构选的是 Redis ZSETkey 是chat:msg:{sessionId}score 就是 seqmember 是包含完整消息内容的 JSON 字符串。为什么不选 List因为补拉的场景是从某个游标往后取增量ZSET 的按分数范围查询天然支持这个场景ZRANGEBYSCORE 一次就能取出来。List 只能从头 pop 或全量 scan要么改数据要么低效都不合适。补拉核心逻辑public ListChatMessage pullMessages(String sessionId, long lastSeq, int limit) { String key chat:msg: sessionId; SetObject raw redisTemplate.opsForZSet() .rangeByScore(key, Range.gt(lastSeq), Limit.limit().count(limit)); // 反序列化并返回 }留存时间设了 24 小时 TTL到期自动过期既能满足“断线多久都能补”的业务需求又不会让 Redis 内存无限增长。如果业务要求更长时间的追溯建议把数据落整套消息数据库ZSET 只做热数据窗口。4.2 两种补拉触发时机连上就补心跳发现落后也要补补拉不能只靠连接建立时那一次因为客户端可能一直在线但消息通道因为某种原因断了一会儿又自动恢复。所以我们设计了两层触发时机。第一层是连接建立后。WebSocket 握手成功后客户端带着本地保存的 lastSeq 调用补拉接口接口返回增量消息和最新 seq。这一层负责处理完整的断线重连。第二层是心跳周期里。客户端和服务端约定每 15 秒一次心跳心跳响应里带上服务端维护的最新 seq。客户端发现最新 seq 比自己本地 lastSeq 大或者发现本地没有某条消息的确认记录就触发一次增量补拉。这一层负责处理静默断流——就是连接还在但消息已经推不过来那种情况。补拉接口设计为POST /api/chat/{sessionId}/messages Body: { afterSeq: 123, limit: 200 } Response: { messages: [...], latestSeq: 222, hasMore: true }这里限流很关键默认每页 200 条客户端通过 hasMore 循环拉取。断线 5 分钟内消息量一般不会超过几十条一页就够极端情况积压上千条时才走分页。4.3 消息幂等客户端和服务端各自管一段补拉和实时推送在重连瞬间一定会出现重叠窗口客户端刚发起补拉请求实时通道的新消息也被推了过来同一条消息就会重复出现。这时候靠的就是幂等机制。服务端的幂等在于消息本身是唯一的。messageId 在消息首次产生时用 UUID 生成后续不管补拉多少次返回的都是同一个 messageId。补拉接口只读不写不会因为重复调用产生新消息。客户端的幂等在于本地维护一个已处理消息 ID 的滑动窗口窗口大小按最近 200 条处理来一条先查窗口存在就丢弃不存在才渲染并把 messageId 加入窗口。同时 seq 小于本地游标的消息直接丢弃因为本地已经渲染过了。这样即使补拉和实时推送同时到达UI 上也不会出现重复气泡。4.4 补拉压测结果与调优从 420ms 降到 90ms方案写完后我们专门做了一轮补拉压测。场景是模拟一个坐席会话持续产生 1000 条消息客户端断线 5 分钟后重连触发全量补拉。第一版测出来补拉接口 P99 耗时 420ms主要瓶颈是返回的数据量太大——1000 条消息 JSON 序列化传输网络和解析开销都很可观。做了三个优化后降到了 90ms 左右。第一分页拉取每页 200 条客户端边拉边渲染第一条用户基本无感。第二ZSET 的 member 从“完整消息 JSON”改成“消息 ID”完整内容按 ID 在另一张缓存里读取减少 ZSET 内存占用。第三补拉接口加 Redis 操作超时 200msRedis 抖动时快速失败不会拖垮整个接口客户端会自动重试。优化后的数据对比如下项目优化前优化后单次补拉 P99 耗时420ms90ms断线 5 分钟消息完整率96.4%99.99%重复消息展示概率较高0客户端窗口去重完整率从 96.4% 提升到 99.99% 的关键不在网络质量而在于 Pub/Sub 丢消息后补拉接口能利用留存数据把缺口全部填回来。这是整个方案价值的直接证明。5. 联调阶段与上线前暴露的问题清单5.1 六个典型问题按表现、根因和解法拆解整个改造从开发和联调持续了两周多遇到的问题不少我整理成一份排雷清单每条都是真实踩过的。问题现象根因解决方案Redis 里消息是二进制乱码序列化器用了 JDK 默认实现全局统一为 Jackson JSON监听器偶尔停止消费处理器抛异常未捕获设置 errorHandler 并在逻辑内 try/catch线程池满后事件丢失拒绝策略用了 AbortPolicy改为 CallerRunsPolicy补拉和实时推送重复缺少恢复正常序的时机客户端滑动窗口去重心跳正常但消息不更新连接处于半开状态增加心跳 seq 对比触发补拉Redis 主从切换时消息丢失Pub/Sub 即发即弃补拉兜底不依赖 Pub/Sub 可靠投递这里面最值得展开的是半开连接问题。TCP 连接看起来还在实际上对端已经不可达服务端不知道客户端挂了客户端还以为连接有效。心跳 seq 对比触发补拉是解决这个问题的最简单办法不用引入复杂的探活机制业务语义本身就带了健康检查。5.2 压测数据200 路真实会话下的表现上线前压测按真实业务场景跑了 30 分钟4 个实例每个实例 2C4G 容器模拟 200 个坐席同时开会话每个会话每分钟约 30 条消息。核心指标如下实时推送通道的端到端延迟 P99 从改造前的 1.2 秒降到 45 毫秒。这里要说明一下P99 降低主要不是变快了而是不再把大模型调用的慢延迟串在推送链路里。AI 回复本身的生成时间还是由模型决定但从事件总线发出到客户端收到这个环节稳定在 45 毫秒以内。补拉接口在断线重连场景下 P99 120 毫秒包含网络往返、Redis 查询、JSON 反序列化全链路。200 路会话同时断线重连时Redis 单实例 CPU 峰值 65%没有出现大 key 热键问题。ZSET 里单会话最大 1000 条消息member 改为消息 ID 后单 ZSET 内存不到 80KB完全可控。5.3 这次收官的三个体会第一个体会是“同步调用在 AI 场景里是最省事但最不可靠的方案”。省事体现在写代码快不可靠体现在只要网络抖一下就原形毕露企业级项目必须把可靠性放在第一位。第二个体会是 Redis 是真的能扛起这种量级的转发。很多人一听到消息队列就上 Kafka其实评估一下吞吐量、延迟要求、运维成本很多场景用 Redis Pub/Sub 加补拉就够。技术选型不是越重越好是够用且可靠最好。第三个体会是断线补拉要早设计、早联调不要在最后上线前才补。我们就是忽略了这一块收尾时加班两周多才赶上节奏。如果从项目一开始就把“消息可靠不丢”作为需求写进迭代后面会轻松非常多。最后分享一个小技巧Pub/Sub 监听器的日志特别重要我们专门在 errorHandler 里加了一个计数器每次异常打日志之外还会记录时间戳。Redis 出故障时看这个日志频率就能快速判断是网络抖动还是代码问题排障效率高不少。客服项目上线后稳了两个月没出大问题回头再看这三件事确实是企业级 AI 客服最核心的地基。