企业官网建设流程全解析

1. 客服系统上线前夜，我们被三个问题卡住了

做过客服系统的人都有一个共识：功能跑通不难，难的是让它敢在真实流量下上线。我们这套基于 Spring AI 的智能客服系统，在测试环境里跑得挺漂亮，单轮问答准确率能到 90% 以上，多轮上下文也能接得住。但临近上线评审，三个问题被反复拎出来：

第一，AI 回复是流式的，前端一个字一个字往外蹦，用户看着爽，可一旦网络抖一下，连接断了，用户那边就卡在半句话上，既没有错误提示，也拿不到完整答案。第二，客服系统不是单机玩具，生产环境是多实例部署，用户这次连到 A 节点，下次可能连到 B 节点，会话上下文怎么保证不丢？第三，也是最要命的，AI 生成一条完整回复可能要好几秒，这期间如果服务重启或者节点扩容，正在生成中的消息直接蒸发，用户等半天等来个寂寞。

这三个问题不解决，客服团队根本不敢把系统推给真实用户。我们最后的方案是：事件总线 + Redis Pub/Sub + 断线补拉。这套组合拳打下来，流式输出的可靠性问题基本被摁住了。下面我把整个设计思路、踩过的坑和最终落地的代码结构，完整拆一遍。

2. 为什么是事件总线，而不是直接调 Service

2.1 直接调用的死结在哪里

最直觉的做法是：Controller 收到请求，直接调 AiService，AiService 调模型，拿到流式结果往 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 还是 Stream？List 简单，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 List<ChatEvent> fetchMissed(String sessionId, long lastSequence) { String streamKey = "ai:chat:stream:" + sessionId; List<MapRecord<String, 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 条消息，超出的从头部淘汰。

// 带长度限制的写入 MapRecord<String, 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 Hash，key 是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 最终沉淀的参数清单

把关键参数整理成表，方便对照调优。

这套方案上线后，客服系统在真实流量下跑了三个月，流式输出的断线续接成功率在 99.9% 以上，用户侧基本感知不到断线。唯一一次异常是 Redis 主从切换，切换期间约 3 秒的消息延迟，但因为补拉机制兜底，用户只感觉到回复慢了一点，没有丢内容。

如果你也在做类似的流式 AI 应用，我的建议是：别等到上线前才想可靠性问题。事件总线和补拉机制应该在架构设计的第一天就纳入考虑，后加虽然也能加，但改造成本高得多。尤其是 sequence 分配和存档写入的顺序，这种细节一旦定错，后面补拉逻辑全得推倒重来。