Spring AI客服项目收尾复盘：事件总线、Redis Pub/Sub与断线补拉实战

项目收尾那天，客服主管跑过来问我：“你们说的 AI 客服助手，用户聊到一半断网再回来，之前的内容能续上吗？”我当时一愣，因为演示环境一直走的是内网，断线续聊这块还真没单独压过。这个问题直接把 Spring AI 客服项目从“把功能跑通”推向了“把消息链路做可靠”，后面两周全部铺在事件总线、Redis Pub/Sub 和断线补拉这三件事上。这篇就是这次企业级收官的完整复盘，包括为什么走这套方案、代码怎么组织、上线前踩了哪些坑，以及补拉压测的真实数字。

1. 客服会话消息为什么会“走着走着就丢了”

1.1 最初架构：WebSocket 直连大模型调用

这个客服项目最初的链路很简单：用户在前端页面发起会话，消息通过 WebSocket 推到后端，后端同步调用大模型，拿到回复后再通过同一个 WebSocket 通道推回给用户。Spring AI 在这里主要负责把大模型接入封装成模板化调用，对业务代码来说就是一个 ChatClient，传 prompt 进去、拿 content 出来。

这套架构在 Demo 阶段完全够用，因为单实例部署、内网环境稳定、消息量也小。但真正往企业级方向收敛时，问题立刻暴露了：WebSocket 连接是长连接，而大模型调用是典型的慢操作，动不动三五秒，高峰期甚至能到十秒以上。同步调用的时候，只要客户端在等待期间断一次网、切一次后台、或者代理超时，这次回复就再也回不到用户屏幕上。用户看到的结果就是“问完问题没反应”，只好再问一遍，然后系统又调一遍大模型，既浪费钱又伤体验。

更麻烦的是，一旦把服务扩展成多实例部署，同步链路会变得根本说不通。用户连的是节点 A，AI 回复如果由节点 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 Map<Class<?>, List<ChatEventHandler>> 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); // 先落库，再分发 List<ChatEventHandler> 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 建立 WebSocket，AI 回复可能由节点 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、timestamp，payload 里至少包含 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/catch：

container.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 ZSET，key 是chat:msg:{sessionId}，score 就是 seq，member 是包含完整消息内容的 JSON 字符串。

为什么不选 List？因为补拉的场景是从某个游标往后取增量，ZSET 的按分数范围查询天然支持这个场景，ZRANGEBYSCORE 一次就能取出来。List 只能从头 pop 或全量 scan，要么改数据要么低效，都不合适。

补拉核心逻辑：

public List<ChatMessage> pullMessages(String sessionId, long lastSeq, int limit) { String key = "chat:msg:" + sessionId; Set<Object> 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 操作超时 200ms，Redis 抖动时快速失败，不会拖垮整个接口，客户端会自动重试。

优化后的数据对比如下：

完整率从 96.4% 提升到 99.99% 的关键不在网络质量，而在于 Pub/Sub 丢消息后，补拉接口能利用留存数据把缺口全部填回来。这是整个方案价值的直接证明。

5. 联调阶段与上线前暴露的问题清单

5.1 六个典型问题，按表现、根因和解法拆解

整个改造从开发和联调持续了两周多，遇到的问题不少，我整理成一份排雷清单，每条都是真实踩过的。

这里面最值得展开的是半开连接问题。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 客服最核心的地基。