1. 客服系统上线前夜我们被三个问题卡住了做过客服系统的人都有一个共识功能跑通不难难的是让它敢在真实流量下上线。我们这套基于 Spring AI 的智能客服系统在测试环境里跑得挺漂亮单轮问答准确率能到 90% 以上多轮上下文也能接得住。但临近上线评审三个问题被反复拎出来第一AI 回复是流式的前端一个字一个字往外蹦用户看着爽可一旦网络抖一下连接断了用户那边就卡在半句话上既没有错误提示也拿不到完整答案。第二客服系统不是单机玩具生产环境是多实例部署用户这次连到 A 节点下次可能连到 B 节点会话上下文怎么保证不丢第三也是最要命的AI 生成一条完整回复可能要好几秒这期间如果服务重启或者节点扩容正在生成中的消息直接蒸发用户等半天等来个寂寞。这三个问题不解决客服团队根本不敢把系统推给真实用户。我们最后的方案是事件总线 Redis Pub/Sub 断线补拉。这套组合拳打下来流式输出的可靠性问题基本被摁住了。下面我把整个设计思路、踩过的坑和最终落地的代码结构完整拆一遍。2. 为什么是事件总线而不是直接调 Service2.1 直接调用的死结在哪里最直觉的做法是Controller 收到请求直接调 AiServiceAiService 调模型拿到流式结果往 response 里写。测试环境这么干没问题但生产环境立刻暴露三个死结。第一个死结是耦合。流式输出的生命周期很长从发起请求到最后一个 token 落地中间可能经历好几秒。这段时间里Controller 的线程被占着AiService 也被占着任何一方想做个超时控制、重试、或者异步落库都得在这条链路上硬塞逻辑越塞越乱。第二个死结是多实例下的会话漂移。用户第一次请求打到节点 A节点 A 在内存里维护了会话上下文。第二次请求因为负载均衡打到节点 B节点 B 内存里没有这个会话AI 就失忆了。你可能会说用 Redis 存会话不就行了对但存进去只是第一步流式生成过程中的中间状态怎么同步生成到一半的 token 序列怎么让其他节点感知第三个死结是生成过程不可恢复。AI 正在生成第 50 个 token这时候节点挂了。用户那边连接断了重新发起请求AI 从头再生成一遍。用户看到的是重复内容体验极差。更糟的是如果这条消息已经落库了一部分重试还会造成数据不一致。2.2 事件总线把生成和消费拆开了事件总线的核心价值是把AI 生成消息这件事从一个同步调用变成一个可观测、可重放、可恢复的事件流。具体来说当用户发起一次对话请求系统不再直接调 AI 然后等结果而是做两件事第一往事件总线发一个对话请求已创建的事件这个事件里带着会话 ID、消息 ID、用户输入等元数据第二AI 生成服务订阅这个事件开始生成每生成一个 token 或者一批 token就往总线发一个消息增量事件。前端那边呢前端不直接跟 AI 生成服务耦合而是订阅这个会话对应的事件流。生成服务发一个增量前端收一个增量渲染一个增量。生成服务发一个生成完成事件前端就知道这条消息结束了。这么一拆好处立刻显现。生成服务和前端消费彻底解耦生成服务挂了重启只要事件还在总线里就能接着消费前端断线重连只要能从总线里补拉断线期间的事件就能把缺失的内容补齐。多实例部署也不再是问题因为会话状态和生成状态都在总线里不在某个节点的内存里。2.3 事件总线的选型为什么最终落在 Redis Pub/Sub事件总线的实现方式有很多Kafka、RabbitMQ、Redis Pub/Sub 都能干。我们最终选 Redis Pub/Sub原因很实际。Kafka 太重运维成本高我们这套客服系统的消息量级还没到需要 Kafka 的程度。RabbitMQ 功能全但引入一个新的中间件意味着多一套运维体系团队精力有限。Redis 我们本来就在用做缓存、做分布式锁再加一个 Pub/Sub 能力几乎零额外成本。但 Redis Pub/Sub 有个众所周知的短板它不持久化消息。发布者发出去的消息如果当时没有订阅者在线消息就丢了。这恰恰是断线补拉要解决的问题。我们的思路是Pub/Sub 负责实时推送保证在线时的低延迟另外用 Redis 的 List 或者 Stream 结构做一份消息存档负责断线后的补拉。两者配合既拿到了 Pub/Sub 的轻量和低延迟又补上了可靠性。这里有个关键决策点存档用 List 还是 StreamList 简单LPUSH LRANGE 就能实现按序读取但没法做消费位点管理。Stream 原生支持消费者组和位点更适合补拉场景。我们最终选了 Stream因为补拉时需要精确知道这个客户端上次读到哪了Stream 的位点机制天然适配。3. Redis Pub/Sub 在流式场景下的真实表现3.1 频道设计一个会话一个频道还是全局一个频道频道设计是第一个要拍板的事。最粗的做法是全局一个频道所有消息都往里发订阅者自己过滤。这么做实现简单但消息量一大每个订阅者都要处理海量无关消息浪费严重。细一点的做法是一个会话一个频道频道名用会话 ID 命名。这样订阅者只订阅自己关心的会话消息精准投递。但问题来了Redis 的频道数量没有硬上限可如果同时有几千个活跃会话就是几千个频道Redis 的订阅管理开销会上升。我们最终采用的是折中方案按会话 ID 做哈希分片比如固定开 64 个频道会话 ID 对 64 取模决定进哪个频道。这样频道数量可控订阅者订阅自己会话对应的那个频道虽然还是会有少量无关消息但量级完全可接受。实测下来单频道每秒几千条消息Redis 毫无压力。// 频道分片逻辑 public String resolveChannel(String sessionId) { int shard Math.abs(sessionId.hashCode()) % 64; return ai:chat:channel: shard; }3.2 消息格式增量事件到底该带哪些字段消息格式设计不好后面补拉和去重都会很痛苦。我们踩过一版坑最初只发了{sessionId, content}结果补拉的时候发现没法去重同一条增量可能被处理两次前端渲染出重复文字。后来定下来的格式是这样的{ eventId: evt_20250101_abc123, sessionId: sess_xyz789, messageId: msg_001, sequence: 42, type: DELTA, content: 您好, timestamp: 1735689600000, finished: false }几个字段的作用得说清楚。eventId是全局唯一 ID用于去重前端和补拉逻辑都靠它判断这条我是不是已经处理过了。sequence是这条消息在本次生成中的序号从 0 开始递增补拉时按序号排序缺哪个补哪个。type区分事件类型有DELTA增量、DONE生成完成、ERROR生成出错几种。finished是个冗余标记方便前端快速判断。这里有个容易忽略的点sequence必须由生成服务在发消息前就分配好不能靠 Redis 的接收顺序。因为 Pub/Sub 不保证跨频道有序甚至同频道在高并发下也可能乱序。自己带序号补拉时才能重建正确顺序。3.3 发布与订阅的代码骨架发布端逻辑不复杂核心是先存档再发布。顺序不能反否则可能出现消息已发布但存档失败补拉时缺数据。public void publish(ChatEvent event) { // 1. 先写入 Stream 存档 String streamKey ai:chat:stream: event.getSessionId(); redisTemplate.opsForStream().add(streamKey, event.toMap()); // 2. 设置过期时间避免存档无限增长 redisTemplate.expire(streamKey, Duration.ofHours(2)); // 3. 再发布到 Pub/Sub 频道 String channel resolveChannel(event.getSessionId()); redisTemplate.convertAndSend(channel, JSON.toJSONString(event)); }订阅端用RedisMessageListenerContainer注册监听器收到消息后先做去重判断再推给前端。Bean RedisMessageListenerContainer listenerContainer(RedisConnectionFactory factory) { RedisMessageListenerContainer container new RedisMessageListenerContainer(); container.setConnectionFactory(factory); // 订阅所有分片频道 for (int i 0; i 64; i) { container.addMessageListener(new ChatMessageListener(), new ChannelTopic(ai:chat:channel: i)); } return container; }去重这块我们用一个本地Caffeine缓存存最近处理过的eventId容量设 10000过期时间 5 分钟。为什么不用 Redis 做去重因为去重判断在消息消费的热路径上走本地缓存延迟低得多而且重复消息通常集中在短时间内本地缓存足够覆盖。4. 断线补拉让用户感觉从未断过4.1 补拉的触发时机与客户端协议补拉不是服务端主动做的而是客户端在重连后主动发起。前端需要维护一个状态我最后成功处理的那条消息的 sequence 是多少。每次收到增量更新这个值。断线重连后前端带着sessionId和lastSequence调一个补拉接口。补拉接口的逻辑很直接从 Stream 里读取该会话的所有事件过滤出sequence lastSequence的按 sequence 排序返回。public ListChatEvent fetchMissed(String sessionId, long lastSequence) { String streamKey ai:chat:stream: sessionId; ListMapRecordString, Object, Object records redisTemplate.opsForStream().range(streamKey, Range.unbounded()); return records.stream() .map(this::toChatEvent) .filter(e - e.getSequence() lastSequence) .sorted(Comparator.comparingLong(ChatEvent::getSequence)) .collect(Collectors.toList()); }这里有个细节补拉返回的是全量增量不是最终结果。前端拿到后按 sequence 顺序把缺失的增量拼接到已有内容后面用户看到的就是无缝续接的效果。4.2 补拉与实时推送的竞态处理补拉和实时推送之间天然存在竞态。场景是这样的前端重连先调补拉接口补拉请求还在路上实时推送的新消息已经到了。如果处理不当可能出现重复渲染或者顺序错乱。我们的处理方式是前端在补拉期间把实时推送的消息先缓存起来不立即渲染。等补拉结果回来先渲染补拉内容再把缓存里的实时消息按 sequence 去重后追加。去重靠eventId缓存里的消息如果 eventId 已经在补拉结果里出现过就丢弃。这个逻辑听起来简单但实现时容易漏掉一个边界补拉结果为空的情况。如果lastSequence已经是最新的补拉返回空列表这时候缓存里的实时消息要全部渲染不能因为补拉为空就跳过。4.3 存档的清理策略别让 Stream 撑爆内存Stream 存档不能无限增长。我们设了两道防线。第一道是过期时间每个会话的 Stream key 设 2 小时过期客服对话通常不会持续超过这个时长。第二道是长度限制用XADD的MAXLEN参数每个 Stream 最多保留 1000 条消息超出的从头部淘汰。// 带长度限制的写入 MapRecordString, Object, Object record MapRecord.create(streamKey, event.toMap()); redisTemplate.opsForStream().add(record, org.springframework.data.redis.connection.stream.XAddOptions.maxlen(1000).approximateTrimming(true));approximateTrimming设为 true 很关键。精确裁剪每次写入都要遍历性能差近似裁剪用~语义Redis 内部按节点批量淘汰性能好得多。代价是实际保留条数可能略多于 1000但对我们的场景完全够用。实测数据单会话 1000 条消息每条平均 200 字节一个会话的存档约 200KB。同时 1000 个活跃会话总内存约 200MB。这个量级对 Redis 来说很轻松但如果你的会话量级是十万级就得考虑把存档挪到更便宜的存储上或者缩短过期时间。5. 多实例部署下的会话一致性怎么保证5.1 会话上下文不能只存内存多实例部署时会话上下文必须外置。我们把会话上下文存 Redis Hashkey 是ai:chat:context:{sessionId}field 是轮次序号value 是该轮的对话内容。每次 AI 生成前先从 Redis 读上下文拼进 prompt生成完再把新的一轮写回去。这里有个并发问题同一个会话如果同时有两个请求进来两个请求都读上下文、都生成、都写回后写的会覆盖先写的。我们的解法是给会话加分布式锁锁的粒度是会话 ID用 Redis 的SET NX PX实现。锁的超时时间设 30 秒比 AI 生成的最长耗时略长。public String acquireLock(String sessionId) { String lockKey ai:chat:lock: sessionId; String token UUID.randomUUID().toString(); Boolean ok redisTemplate.opsForValue() .setIfAbsent(lockKey, token, Duration.ofSeconds(30)); return Boolean.TRUE.equals(ok) ? token : null; }释放锁时要用 Lua 脚本校验 token防止误删别人的锁。这是分布式锁的标准操作不展开说了。5.2 生成服务的无状态化改造生成服务本身要尽量无状态。所有状态——会话上下文、生成进度、消息存档——全部外置到 Redis。生成服务实例可以随时启停不影响正在进行的会话。具体来说生成服务收到对话请求已创建事件后做这几步第一步获取会话锁第二步从 Redis 读上下文第三步调 AI 模型流式生成第四步每生成一批 token分配 sequence发增量事件第五步生成完成发 DONE 事件写回上下文释放锁。如果生成到一半服务挂了锁会因为超时自动释放上下文没有被写回因为写回在最后一步所以下次请求会基于旧上下文重新生成。这会导致用户看到重复内容吗不会因为前端有 sequence 去重而且重新生成的消息会分配新的 messageId前端按 messageId 区分不同轮次。5.3 节点扩容时的平滑过渡节点扩容时新节点启动开始订阅 Pub/Sub 频道。但这里有个问题新节点订阅之前频道上已经有一些消息在流转新节点收不到。不过没关系因为新节点不负责补拉补拉是客户端主动发起的客户端会从 Stream 里读跟节点无关。真正需要注意的是订阅关系的一致性。如果每个节点都订阅全部 64 个频道那所有节点都会收到所有消息然后各自判断这条消息是不是我的会话。这么做简单但浪费。我们的做法是让节点只订阅自己负责的频道分片通过一致性哈希把会话分配到节点。不过考虑到客服系统的会话量级和节点数我们最终没做这层优化直接全量订阅靠会话 ID 过滤。实测下来64 个频道的全量订阅单节点 CPU 开销可以忽略。6. 踩过的坑和最终沉淀下来的经验6.1 坑一Pub/Sub 消息丢失不是偶尔是必然最初我们以为 Redis Pub/Sub 丢消息是小概率事件结果压测时发现只要订阅端稍微卡一下比如 GC 停顿消息就丢。因为 Pub/Sub 是 fire-and-forget发布者发完就不管了订阅者没收到就是没收到。这个坑让我们彻底放弃了只靠 Pub/Sub的幻想老老实实加了 Stream 存档和补拉。现在回头看Pub/Sub 的定位应该是实时加速通道而不是可靠传输通道。可靠传输必须靠存档 补拉。6.2 坑二sequence 分配不能放在发布之后我们最初的设计是生成服务生成一批 token先发 Pub/Sub再分配 sequence 写 Stream。结果补拉时发现Stream 里的 sequence 顺序和 Pub/Sub 的到达顺序对不上前端补拉后拼接的内容错乱。根因是 Pub/Sub 和 Stream 是两条独立的写入路径没有事务保证。修复方案很简单先分配 sequence再写 Stream最后发 Pub/Sub。sequence 在生成服务内部用一个AtomicLong维护每个 messageId 一个计数器。6.3 坑三补拉接口的鉴权不能省补拉接口带着 sessionId 就能拉取该会话的全部消息这本身是个安全隐患。如果 sessionId 泄露别人就能拉到完整对话。我们的补救措施是补拉接口除了 sessionId还要校验一个clientToken这个 token 在会话创建时下发给客户端存在 Redis 里补拉时比对。public boolean validateClient(String sessionId, String clientToken) { String key ai:chat:client: sessionId; String stored redisTemplate.opsForValue().get(key); return clientToken ! null clientToken.equals(stored); }这个 token 跟用户身份是两回事它只代表这个客户端有权限拉取这个会话的消息。用户身份校验在更上层做这里只做会话级别的隔离。6.4 坑四前端渲染增量时的半字符问题流式输出按 token 切分但 token 不等于字符。一个中文字可能被切成两个 token前端如果按 token 直接渲染会出现半个字。我们的解法是前端维护一个缓冲区收到增量后先追加到缓冲区然后按完整字符边界渲染。具体实现依赖前端的字符处理库这里不展开但这个问题在中文场景下非常普遍必须提前考虑。6.5 最终沉淀的参数清单把关键参数整理成表方便对照调优。参数取值说明频道分片数64会话 ID 哈希取模Stream 过期时间2 小时覆盖单次客服对话时长Stream 最大长度1000 条近似裁剪实际略多会话锁超时30 秒略长于 AI 最长生成耗时去重缓存容量10000 条本地 Caffeine去重缓存过期5 分钟覆盖重复消息集中期补拉批次大小500 条避免单次返回过大这套方案上线后客服系统在真实流量下跑了三个月流式输出的断线续接成功率在 99.9% 以上用户侧基本感知不到断线。唯一一次异常是 Redis 主从切换切换期间约 3 秒的消息延迟但因为补拉机制兜底用户只感觉到回复慢了一点没有丢内容。如果你也在做类似的流式 AI 应用我的建议是别等到上线前才想可靠性问题。事件总线和补拉机制应该在架构设计的第一天就纳入考虑后加虽然也能加但改造成本高得多。尤其是 sequence 分配和存档写入的顺序这种细节一旦定错后面补拉逻辑全得推倒重来。