团队做智能客服到上线前夜最慌的不是模型效果不够好而是消息在多个服务节点之间流转的时候一个网络抖动就能让整条链路悬在半空。同步调用会堵、单机内存事件会丢、Redis Pub/Sub 断线会漏消息。所以这次Spring AI 企业级收官核心就是把事件总线、Redis Pub/Sub、断线补拉三件事做扎实让客服系统终于敢放开流量。这篇文章就围绕这三块讲清楚为什么这么做、代码怎么落地、踩过哪些坑。1. 项目背景与设计思路拆解1.1 智能客服上线前的真实焦虑先说说背景。我们团队做了一个基于 Spring AI 的智能客服系统接入大模型后简单 demo 很快就跑通了用户发一句消息系统返回一段回答体验还挺好。但真正到了企业级上线前我连续几个晚上睡不着。原因很简单demo 里的 AI 对话流程是同步的、单机的而生产环境里有多个服务实例、有网络抖动、有高并发随便哪个环节出问题用户看到的都是客服失联。标题里提到的三件事就是最终决定上线的三个关键基础设施。它们要解决的问题非常明确事件总线让客服系统内部各个模块不再互相阻塞AI 生成慢的时候接入层不用干等。Redis Pub/Sub让多实例之间能够协同处理消息不管用户请求落在哪个节点其他节点都能感知到上下文变化。断线补拉在任何断线场景下消息都不丢Redis 断连、应用重启、WebSocket 断开都有兜底方案。这三样缺一样客服系统都不敢放量。1.2 整体架构的分层逻辑整体架构可以按数据流向拆成四层。接入层负责用户消息的接收支持 WebSocket 和 HTTP 两种方式会话层维护每个用户的对话上下文与状态机AI 层通过 Spring AI 的 ChatClient 与大模型交互支持流式输出事件层则承载了所有模块之间的异步通信包括内部事件总线、Redis Pub/Sub 和断线补拉的补偿机制。这个分层里最容易被忽略的是AI 调用慢这件事对整个系统的影响。一次大模型调用可能耗时几秒到几十秒如果接入层同步等待 AI 结果一个用户就可以占住一个连接线程十几个用户就能把线程池耗光。所以必须把 AI 调用过程拆开异步化、事件化让接入层收到消息后立即响应后续结果通过事件机制回调。1.3 为什么没直接上重型消息队列很多团队遇到多实例消息分发第一反应是引入 MQ。我们当时也认真考虑过但最终决定先用 Redis Pub/Sub。原因很实际团队已经有 Redis 运维经验不需要额外部署集群Pub/Sub 的语义足够满足当前场景而且它能天然复用已有的 Redis 连接。虽然 Pub/Sub 有断线丢消息的弱点但我们用断线补拉机制弥补了这个短板整体可靠性可以接受。这个选型也符合渐进式演进的思路。如果后续消息量级上去了、路由规则复杂了再平滑迁移到 MQ 也不迟但当前阶段不为了技术选型而增加基础设施负担。2. 事件总线内部解耦的基石2.1 同步调用撑不住的三个场景事件总线解决的是模块间的强耦合问题。典型场景我列了三个第一个是用户消息进来后会话管理器需要通知 AI 处理器去调用大模型同时又要通知会话状态机更新状态如果这些调用是同步的那整个链路就绑在 AI 调用的耗时上。第二个是 AI 回复生成完成后需要同时触发消息推送、会话持久化、上下文归档等多个后续动作。如果这些后续动作用同步代码写在一个方法里任何一个环节出问题都会导致回复迟迟推不到用户端。第三个是超时和重试的逻辑。AI 调用偶发超时是常态如果同步编程重试逻辑会侵入业务代码加一个重试就多一堆 if else。事件总线的价值就是把发生了什么和怎么处理解耦。系统里每个人只关心自己感兴趣的事件一个事件可以被多个处理器监听多个事件也可以被同一个处理器统一处理。2.2 自研轻量事件总线的设计我们没有引入额外的框架而是在 Spring 的 ApplicationEventPublisher 基础上做了一层封装核心是三个扩展点。第一是同步转异步。Spring 默认的 EventListener 是同步执行的我们改成通过自定义线程池执行避免事件监听器阻塞发布线程。需要说明的是这里要注意线程池的隔离AI 调用耗时较长不能用默认的公共线程池要单独建一个事件处理线程池核心线程数和最大线程数都单独配置。第二是事件携带上下文。自研的 BaseEvent 里包含 eventId、occurredAt、traceId 和 sourceType。traceId 特别重要一个用户会话的所有事件都携带同一个 traceId排查问题时可以直接按 traceId 拉全链路日志。第三是支持事件拦截器。可以自定义实现事件发布前后的钩子用于记录事件耗时、统计事件量、甚至做灰度开关。这个功能在排查线上问题时帮了大忙。public class BaseEvent { private String eventId; private Long occurredAt; private String traceId; private String sourceType; // getter/setter 省略 }2.3 事件定义的颗粒度事件颗粒度决定了事件总线的可用性。太粗的事件会让监听器被迫做很多判断太细的事件又会让系统变得碎片化。我们最终沉淀了四个核心事件CustomerMessageReceivedEvent用户消息接入、AISessionStartedEventAI 会话开始、AIResponseGeneratedEventAI 回复生成完成、AIConnectionLostEventAI 连接断开触发补拉流程。每个事件都有明确的发布时机和监听方。比如 CustomerMessageReceivedEvent 由接入层发布会话管理器和 AI 处理器都监听它AIResponseGeneratedEvent 由 AI 处理器发布消息推送器和持久化器都监听它。这种模型下新增一个功能只需要新增监听器不需要改动发布方的代码。3. Redis Pub/Sub多实例协同的消息通道3.1 为什么需要跨实例分发事件总线解决了单机内模块解耦但还有一个问题系统有多个客服节点实例用户请求经过负载均衡可能落在任意节点上。某个节点上的 AI 处理器生成了回复另一台节点上的消息推送器需要感知到这个事件并推送给用户这就需要一条跨实例的消息通道。Redis Pub/Sub 在这里承担的就是这个职责。它天然支持发布订阅模型发布者把消息发到某个频道所有订阅了这个频道的节点都能收到消息。这样任何一个实例上发生的事件都能广播到集群里所有实例。需要特别说明的是我们是事件总线 Redis Pub/Sub配合使用内部事件先用本地事件总线处理如果这个事件需要被其他实例感知再通过事件总线里配置的转发器把它发布到 Redis 频道。这样内部逻辑和跨实例逻辑各自清晰不会混在一起。3.2 频道设计与消息格式频道设计直接影响可维护性。我们没有为每个业务类型单独建一个频道而是按事件大类建了三个频道channel.customer.message用户消息相关事件channel.ai.sessionAI 会话生命周期事件channel.system.alert系统异常与补偿触发事件频道不宜过多否则订阅连接数量会膨胀Redis 的 Pub/Sub 是逐连接推送的订阅连接越多Redis 压力越大。消息格式我们统一用 JSON。这里有一个关键细节不同实例之间必须保证事件对象的序列化/反序列化兼容性。我们要求所有跨实例传输的事件必须显式声明 schemaVersion避免一个实例升级后发布的对象另一个旧实例解析不了。{ eventId: uuid, eventType: AIResponseGeneratedEvent, schemaVersion: 1, traceId: trace-uuid, payload: { sessionId: session-001, content: 你好我是客服助手, tokens: 128 } }3.3 Redis 连接的 HWM 与线程模型Redis Pub/Sub 的消费者端有一个坑就是用默认的 JdkSerializationRedisSerializer 序列化会导致对象结构绑定 Java 类名跨实例解析时容易出问题而且序列化体积很大。我们全部换成了 Jackson 的 GenericJackson2JsonRedisSerializer并且自定义 ObjectMapper启用 default typing 但限定白名单包路径。再一个关键点是订阅监听器的线程模型。RedisMessageListenerContainer 默认把收到的每条消息分派给监听器线程执行。如果监听器里处理逻辑耗时过长比如去调 AI 接口就会出现消息积压后续消息延迟越来越大。我们的做法是监听器里只做快速的转发和解码真正的业务处理再丢到业务线程池异步执行。Bean public RedisMessageListenerContainer redisMessageListenerContainer( RedisConnectionFactory connectionFactory, MessageListenerAdapter messageListenerAdapter) { RedisMessageListenerContainer container new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.addMessageListener(messageListenerAdapter, PatternTopic.of(channel.*)); container.setTaskExecutor(redisListenerExecutor()); return container; }4. 断线补拉消息可靠性的最后一道防线4.1 Redis Pub/Sub 的致命短板Redis Pub/Sub 最大的问题是消息不持久化。发布者把消息发出来如果此时某个消费者断开连接或者消费者连接还在但处理失败这条消息就永远丢了。网络抖动、应用重启、Redis 主从切换、连接空闲超时任何一个场景都可能导致消息丢失。对客服系统来说消息丢失意味着用户发了一条消息但客服根本没收到或者 AI 生成了回复但用户看不到。这在企业级场景不可接受。所以我们必须设计断线补拉机制这也是标题里终于敢上线的关键所在。断线补拉的核心思想是消息不依赖实时推送作为唯一通道每个节点把已发布的事件消息追加写入本地事件仓库当发现 Redis 连接断开后恢复连接时从事件仓库按游标补拉未处理的消息。4.2 本地事件仓库设计本地事件仓库本质上是一个并发安全的消费者端消息日志记录每个实例已经消费过的事件消息。我们选择了 RocksDB 作为事件仓库存储而不是直接用 Redis 或数据库。RocksDB 是嵌入式 KV 存储本地落盘、写入吞吐高、不依赖外部服务很适合作为断线补拉的游标存储。每个事件消息写入事件仓库时分配一个单调递增的本地序号同时记录 Redis 消息的全局标识。消费端在处理 Redis 消息时先检查全局标识是否已经在仓库里如果已存在说明是重复消息直接跳过如果不存在则写入仓库并处理业务逻辑。补拉流程是这样的节点检测到与 Redis 的连接恢复后从本地仓库中记录的最后消费序号开始向 Redis 发送一个补拉请求。这个请求会触发 Redis 端的一个专门生产者重新发布最近 N 秒内的关键事件消息。实际上这个重新发布需要我们维护一个最近事件的缓存队列保存最近 5 分钟内的消息副本这样即使 Redis Pub/Sub 丢消息了也能从副本中恢复。4.3 断线检测与补拉时机断线检测不能完全依赖 Redis 底层的连接状态回调。我们做了三重检测机制第一重是 RedisConnectionFailureListenerRedis 连接断开时立即感知标记断线状态。第二重是心跳哨兵。单独起一个定时任务每 5 秒向 Redis 发布一条 heartbeat 消息到channel.system.alert频道。每个节点都会收到所有节点的 heartbeat消息里带节点ID如果一个节点超过 3 个心跳周期没被其他节点听到就判定断线。第三重是消费延迟监控。统计 RedisMessageListenerContainer 的消息处理延迟如果延迟超过阈值则认为消费者消费能力出现问题提前启动补拉准备。补拉触发时机也有讲究。不是一恢复连接就立即补拉那样会造成重连风暴。我们的策略是恢复连接后统一等待 2 到 3 秒让 Redis 连接稳定然后按照本地仓库游标一次性补拉所有缺失消息。4.4 消息幂等处理补拉机制引入后最直接的副作用是重复消息暴增。同一个事件可能在实时通道收到一次、补拉通道又收到一次甚至多次。所以幂等处理是必须的。幂等方案用 Redis 分布式锁 本地去重表双重保证。每个事件消息有一个全局唯一 ID处理前先尝试在 Redis 中写入这个 ID 的标记利用 SETNX 命令保证只有一个节点能处理同一条消息。同时本地事件仓库也维护一个已经处理消息的 ID 集合双保险防止重复。这里给一个建议幂等键的范围不能太粗比如不能用 traceId 做幂等键因为一个 traceId 下有多条消息也不能太细比如只用事件 ID那重试机制就无法覆盖处理一半失败的情况。我们用的是事件 ID 事件类型 会话 ID共同组成幂等键。5. 核心代码实现与关键参数配置5.1 事件总线发布端实现先看发布端的实现。在 BusinessEventPublisher 中核心是把同步事件发布和异步事件发布分开同时支持转发到 Redis。Component public class BusinessEventPublisher { private final ApplicationEventPublisher applicationEventPublisher; private final RedisTemplateString, Object redisTemplate; private final EventForwardProperties forwardProperties; private final ThreadPoolTaskExecutor eventExecutor; public void publish(BaseEvent event) { event.setEventId(UUID.randomUUID().toString()); event.setOccurredAt(System.currentTimeMillis()); // 本地事件总线异步派发 eventExecutor.execute(() - applicationEventPublisher.publishEvent(event)); // 如果需要跨实例感知转发到 Redis if (forwardProperties.getForwardSet().contains(event.getSourceType())) { String channel resolveChannel(event); redisTemplate.convertAndSend(channel, event); } } private String resolveChannel(BaseEvent event) { if (event instanceof CustomerMessageReceivedEvent) return channel.customer.message; if (event instanceof AISessionStartedEvent) return channel.ai.session; return channel.system.alert; } }发布端注意两点。第一eventExecutor 的拒绝策略必须设置为 CallerRunsPolicy避免线程池满时直接把事件丢弃。虽然 CallerRunsPolicy 会在调用线程执行但总比丢消息强。第二转发到 Redis 的操作不能影响本地事件发布这里转发是在同一个线程内顺序执行的如果 Redis 不可用要捕获异常并记录到本地异常事件表等待补拉流程处理。5.2 消费者端监听实现消费者端核心是 MessageListenerAdapter 里设置消息转换器和处理器。Bean public MessageListenerAdapter messageListenerAdapter(EventMessageHandler handler) { MessageListenerAdapter adapter new MessageListenerAdapter(handler, handleMessage); adapter.setSerializer(jacksonSerializer()); return adapter; } Component public class EventMessageHandler { private final EventOutboxRepository outboxRepository; private final BusinessEventPublisher eventPublisher; public void handleMessage(String message) { BaseEvent event JSON.parseObject(message, BaseEvent.class); String idempotentKey buildIdempotentKey(event); // 幂等检查 if (outboxRepository.exist(idempotentKey)) { return; } // 先写入本地仓库再处理业务 outboxRepository.append(event, idempotentKey); eventPublisher.publish(event); } }这里有一个顺序细节先写本地事件仓库再执行业务逻辑。为什么因为如果先执行业务逻辑再写仓库业务执行过程中应用宕机了这条消息就既没被业务处理完仓库里也没有记录补拉时就无法感知。先写仓库虽然可能导致业务还没处理但仓库已记录但因为后续还有幂等兜底最多是重复处理不会漏处理。5.3 断线补拉核心实现断线补拉的核心是本地事件仓库的游标管理和 Redis 端的补拉请求处理。游标管理我们用 RocksDB 指定 column family 存储key 是会话 ID 加本地序号value 是消息 JSON。补拉触发代码Component public class CatchUpManager { private final RedisConnectionFactory connectionFactory; private final OutboxRepository outboxRepository; private final RecentlyEventCache recentEventCache; public void catchUpAfterReconnect() { // 连接恢复后等待稳定 Thread.sleep(2000); long lastSequence outboxRepository.getLastSequence(); ListBaseEvent missedEvents recentEventCache.getEventsAfter(lastSequence); for (BaseEvent event : missedEvents) { // 重新投递到本地事件总线 eventPublisher.publish(event); } } }实际实现中recentEventCache 维护的是一个环形缓冲保存最近 5 分钟的所有跨实例事件消息。这样断线不超过 5 分钟时可以通过补拉完全恢复超过 5 分钟的极端情况会触发全量会话状态重建流程从持久化存储恢复。5.4 关键参数的配置经验线程池参数是第一个重点。事件处理线程池 corePoolSize 设置为 8maxPoolSize 设置为 32队列容量设置为 512。这个配置在不同场景下差异很大我们压测下来发现AI 生成事件是高耗时任务单事件平均耗时 2 秒左右如果 32 个线程同时处理每秒吞吐约 16 个事件能满足客服场景的并发量。队列容量设置不要太大否则线程池一旦饱和消息延迟会急剧恶化。宁可触发拒绝策略走补拉也不要让消息在队列里积压到超时。Redis 连接参数也要调整。Pub/Sub 场景下RedisConnectionFactory 需要设置合理的 readTimeout默认的 2 秒在消息量大时容易误判超时。我们设置为 5 秒同时把 TCP keepalive 打开让底层连接能快速感知网络异常。还有一个容易被忽略的参数是 RedisMessageListenerContainer 的 subscription 的恢复策略。我们遇到过一个情况某次 Redis 短暂不可用恢复后监听容器没有自动重新订阅导致频道消息一直收不到。解决方法是给容器设置一个专门的恢复监听器在 connection 恢复后调用container.start()重新订阅所有频道。6. 踩坑记录与问题排查实录6.1 高频踩坑汇总上线前后我们记录了不少问题这里挑几个典型的有代表性的分享。问题现象根本原因解决方案Redis 断线恢复后消息停止接收监听容器未自动重新订阅实现恢复监听器检测连接恢复后重新 start 容器补拉后业务重复执行幂等键范围过粗改为事件 ID 事件类型 会话 ID 组合键消息序列化异常默认 JDK 序列化绑定类名改用 Jackson 白名单反序列化消息延迟随并发增加而飙升监听器线程内直接执行业务逻辑监听器只做解码和转发丢业务线程池多个节点同时补拉导致 Redis 过载补拉时机无协调统一等待 2 到 3 秒 分布式锁控制补拉节点6.2 一次真实的断线丢消息排查过程上线后遇到过最诡异的一次问题某个支付咨询会话中用户连续发了两条消息系统只回复了第一条第二条完全没响应。日志里没有异常链路 traceId 也能搜到部分日志但就是缺一段。排查过程走了不少弯路最后用三把钥匙解开。第一把钥匙看 Redis 日志发现那个时间段发生过一次主从切换Redis 连接断开了约 4 秒。第二把钥匙看本地事件仓库发现第一条消息的消费记录存在第二条消息完全没入库。第三把钥匙看补拉逻辑发现补拉触发的前提是心跳哨兵检测到断线而主从切换时间只有 4 秒心跳周期是 5 秒所以 3 个周期没收到心跳的阈值根本没触发补拉流程没有启动。这个问题的根因在于断线检测的心跳周期粒度大于实际断线时间导致短时间断线没有被感知到。解决方法是增加一个Redis 连接事件监听的快速检测通道监听 RedisConnectionFactory 的连接创建和销毁事件实时感知连接状态。同时把心跳检测阈值从 3 个周期缩短到 2 个周期并降低心跳周期到 3 秒增加了检测灵敏度。6.3 补拉风暴与重连风暴的避免补拉机制如果设计不当会在故障恢复时造成二次故障。我们遇到过最极端的情况是Redis 抖动后所有节点同时触发补拉补拉请求瞬间打满 Redis 的 CPU导致服务长时间不可用。现在的做法是把补拉请求加分布式锁在任意时刻只允许一个节点执行补拉操作。其他节点发现锁被占用就延长等待时间。补拉完成后通过 Redis Pub/Sub 广播一个补拉完成消息通知其他节点可以从缓存中获取补拉结果而不需要再次拉取。另外要控制补拉速率。即使有锁一次性补拉 1000 条消息也会对系统产生压力。我们在补拉循环里加入了速率限制每批最多处理 50 条每批之间间隔 100 毫秒。这样即使消息很多也能平滑恢复。6.4 实操心得日志与监控是最后一张安全网最后这部分不是代码层面的但我觉得比代码更重要。客服系统的消息链路长、节点多、异常隐蔽没有好的日志和监控排查问题就像大海捞针。我们做了一套事件链路日志规范每个事件发布时写入 INFO 日志包含事件 ID、traceId、事件类型消费者接收时也写入同样的结构化日志。这样通过 traceId 就能串联完整链路。监控上我们重点看三个指标事件总线线程池活跃线程数、Redis Pub/Sub 消息积压延迟、断线补拉事件数量和耗时。任何一个指标异常都会触发告警这样故障往往在用户感知前就已经被发现和处理了。7. 最后的几点经验总结回到标题说句实话Spring AI 本身做智能客服并不难难的是让它企业级可用。事件总线解决的是系统内部的耦合问题Redis Pub/Sub 解决的是跨实例分发问题断线补拉解决的是消息可靠性问题。这三者组合起来构成了一个既有解耦能力又有兜底方案的完整体系。如果让我重新做一遍这个系统我会在一开始就先把消息可靠性的地基打好而不是先跑通 demo 再补。因为断线补拉不是事后想加就能轻松加的它要求所有事件在发布时就有全局唯一 ID、在消费时有幂等机制、在本地有可回溯的事件仓库。这些设计必须从第一天就融入代码里否则后期改造会伤筋动骨。另外Redis Pub/Sub 虽然轻量但它的可靠性短板是客观存在的。如果你所在团队有使用 MQ 的条件并且消息可靠性的要求非常高那直接选 MQ 可能更省心。但如果像我们一样希望保持基础设施精简那么Pub/Sub 断线补拉这套组合在客服场景下是够用的前提是把补拉机制真正做透而不是把它当摆设。最后再分享一个小技巧补拉机制里的事件仓库最好做成可查询的。我们后来遇到一个棘手问题用户说我早上发了一条消息客服回复我没收到就是靠查本地事件仓库找到那条消息的处理状态和流向最终定位到是推送环节的连接问题。这个仓库不仅为消息可靠服务也为日后的审计和问题定位提供了依据。价值远比表面上看起来要大。
