企业官网建设流程全解析

LangChain4j 企业级收官：事件总线 + Redis Pub/Sub + 断线补拉，客服终于敢上线

作者：鱼宵 ｜ LangChain4j 实战精通营 · 第 13 篇

上一课我们做出了"逐字流式回答 + 思考面板 + 工具调用徽章"的客服，我本以为这下稳了。结果上线演练第一天就被运维泼冷水：你这 SSE 事件是业务层直接拿SseEmitter发的，单实例是爽，多实例怎么搞？浏览器连在 A 上、请求打到 B，B 根本拿不到 A 上的 emitter，事件往哪推？

我当时愣了一下。再一盘点，这层自研事件链有四个死穴：业务层和传输层耦死、事件没统一 schema、断线就丢事件、也没有限流重试和指标。说白了，它就是个"单店后厨直接端菜上桌"的 demo——只在这一家店成立。这篇就把这层重做成企业级方案，也是这个系列的收官篇。代码全在仓库的lesson-13/目录，clone 下来照着命令起两个实例，你能亲眼看到"浏览器连 A、请求打 B，事件照样流回浏览器"。

一、核心原理：单店端菜为什么上不了连锁

先认清 L12 事件链的四个死穴：

1. 业务层和传输层耦死：Service 里 new 了 SSE 对象，想换 WebSocket、想加事件落盘、想多实例转发，全得改业务代码。
2. 事件没 schema：thinking 是个 Map、token 是裸字符串、done 是另一个 Map，前端解析全靠猜。
3. 没有 id：浏览器刷新/断网重连，不知道从哪条补起，直接丢一段回答。
4. 单实例绑死：SseEmitter是 JVM 内存对象，多实例下事件发不回去。

类比：L12 像"后厨直接把菜端到你桌上"——只在这一家店成立。开了分店，你在 A 店入座、菜在 B 店炒出来，B 店服务员没你这桌的牌子，菜端不出来。

解法分三步。第一步，把所有事件统一成一个record AgentEvent（带编号 id、类型 type、时间 ts、归哪个会话 sessionId、同一次提问 requestId、traceId、结构化 payload）。没 id 不能断线补拉，没 schema 前后端没法协作。

第二步，用Spring 事件总线解耦：业务层只管publishEvent(AgentEvent)，容器回调@EventListener，配@Async就在异步线程池跑，publish 立刻返回不阻塞业务。> 报社热线类比：记者（业务层）打完电话就去写稿，接线员（监听器）在另一间屋把稿子塞进"传菜电梯"（Redis）。记者绝不亲自送报。

第三步，用Redis Pub/Sub 做事件网关：监听器拿到事件，RedisINCR分配全局单调 id，序列化成 JSON，convertAndSend到"每会话独立频道"lc4j-l13:events:{sessionId}，同时RPUSH进回放 List 留最近 100 条。每个实例都订阅这个频道，收到后看本地 Map 里有没有这个 sessionId 的 emitter——有就转发给浏览器，没有就丢。

跨实例铁证：浏览器 SSE 连在 A，请求打到 B 产生事件，B 发到 Redis，Redis 扇出给所有实例，A 听到后转给浏览器。事件从 B 出发、经 Redis、由 A 上桌。

生产加固四件套：按 sessionId 动态限流（Resilience4j，每会话 10 次/秒，超了返 429）、模型调用重试+熔断（连续失败率超 50% 跳闸 10 秒走兜底）、Redis 做 FAQ 确定性缓存（归一化后命中直接回标准答案，零 token 成本）、会话 token 鉴权。

二、动手：起两个中间件，再跑两个实例

本课 Redis 在6391（容器redis-l13）、Zipkin 在9412（容器zipkin-sa），应用实例 A 端口10004、实例 B 端口10006。

第 0 步：确认容器在跑。

dockerps--filter"name=redis-l13"--filter"name=zipkin-sa"# 都是 Up 才继续java-version[bool][Environment]::GetEnvironmentVariable('DEEPSEEK_API_KEY')

第 1 步：编译打包。

cd langchain4j-journey\lesson-13$env:JAVA_HOME="C:\Program Files\Java\jdk-17"mvn clean install-DskipTests# BUILD SUCCESS

第 2 步：起实例 A（前台跑，能看日志）。

java-jar target\lesson-13-1.0.0.jar--server.port=10004

看到[FAQ缓存] 预载完成：... 12 条 FAQ和Tomcat started on port 10004即成功。

第 3 步：浏览器开前端。http://localhost:10004/，页面会自动建会话、连 SSE。

第 4 步（见证横向扩展）：再起实例 B。新开一个 PowerShell：

java-jar target\lesson-13-1.0.0.jar--server.port=10006

浏览器 SSE 连在 10004，但你用 curl 把请求打到 10006——事件照样在浏览器里流出来（第四节有实测）。

三、关键代码：事件信封 + Redis 网关 + 本地转发

① 统一事件信封 AgentEvent（id 不在业务层赋值，交给 Redis INCR）。

packagecom.langchain4j.lesson13.event;importjava.util.Map;/** 统一事件信封：一次"发生了什么"的标准化记录。 */publicrecordAgentEvent(longid,// 每会话单调递增（Redis INCR 分配），断线重连前端带 Last-Event-ID 补拉Typetype,// 枚举：thinking/tool/token/citations/done/error/rate_limitedintversion,// 协议版本，将来改字段不炸老前端longts,// 毫秒时间戳，前端画时间线StringsessionId,// 路由到哪个会话的 SSEStringrequestId,// 同一次提问的事件归到同一个气泡StringtraceId,// 链路号，进 payload，前端/日志/Zipkin 三处可对Map<String,Object>payload){publicenumType{thinking,tool,token,citations,done,error,rate_limited}// 业务层构造"还没编号"的事件；id 由 Redis INCR 在监听器里分配（跨实例单调，A/B 不撞号）publicstaticAgentEventof(Typetype,Stringsid,StringreqId,StringtraceId,Map<String,Object>p){returnnewAgentEvent(0L,type,1,System.currentTimeMillis(),sid,reqId,traceId,p);}}

id 故意不在业务层赋值——跨实例要全局单调递增，必须交给 RedisINCR这个"发号器"，否则 A、B 各自计数会撞号。

② 事件网关出口：一个事件干两件事——广播 + 落回放。

packagecom.langchain4j.lesson13.event;importcom.fasterxml.jackson.databind.ObjectMapper;importorg.springframework.context.event.EventListener;importorg.springframework.scheduling.annotation.Async;importorg.springframework.data.redis.core.StringRedisTemplate;importorg.springframework.stereotype.Component;@ComponentpublicclassAgentEventListener{privatefinalStringRedisTemplateredis;privatefinalObjectMapperjson;publicAgentEventListener(StringRedisTemplateredis,ObjectMapperjson){this.redis=redis;this.json=json;}@Async// 异步线程池跑：业务 publish 完立刻继续，不等 Redis、不等 SSE@EventListenerpublicvoidon(AgentEventevent)throwsException{Longid=redis.opsForValue().increment("lc4j-l13:seq:"+event.sessionId());// 发号Stringbody=json.writeValueAsString(event.withId(id==null?0L:id));redis.convertAndSend("lc4j-l13:events:"+event.sessionId(),body);// ① 扇出广播给在线连接redis.opsForList().rightPush("lc4j-l13:replay:"+event.sessionId(),body);// ② 落回放 Listredis.opsForList().trim("lc4j-l13:replay:"+event.sessionId(),-100,-1);// ③ 只留最近 100 条}}

③ SSE 桥：本地有 emitter 才转发，断线重连补拉。

packagecom.langchain4j.lesson13.web;importcom.fasterxml.jackson.databind.ObjectMapper;importcom.langchain4j.lesson13.event.AgentEvent;importcom.langchain4j.lesson13.event.AgentEventListener;importorg.springframework.data.redis.connection.Message;importorg.springframework.data.redis.connection.MessageListener;importorg.springframework.data.redis.core.StringRedisTemplate;importorg.springframework.stereotype.Component;importorg.springframework.web.servlet.mvc.method.annotation.SseEmitter;importjava.nio.charset.StandardCharsets;importjava.util.Map;importjava.util.concurrent.ConcurrentHashMap;@ComponentpublicclassSseEventBridgeimplementsMessageListener{privatefinalMap<String,SseEmitter>emitters=newConcurrentHashMap<>();// 本实例的 SSE 连接privatefinalStringRedisTemplateredis;privatefinalObjectMapperjson;publicSseEventBridge(StringRedisTemplateredis,ObjectMapperjson){this.redis=redis;this.json=json;}@OverridepublicvoidonMessage(Messagemessage,byte[]pattern){Stringchannel=newString(message.getChannel(),StandardCharsets.UTF_8);Stringbody=newString(message.getBody(),StandardCharsets.UTF_8);if(!channel.startsWith(AgentEventListener.CHANNEL_PREFIX))return;StringsessionId=channel.substring(AgentEventListener.CHANNEL_PREFIX.length());SseEmitteremitter=emitters.get(sessionId);// 本实例没这条 SSE → 丢掉if(emitter==null)return;try{AgentEventev=json.readValue(body,AgentEvent.class);emitter.send(SseEmitter.event().id(String.valueOf(ev.id()))// SSE 帧的 id，浏览器重连自动当 Last-Event-ID 带回.name(ev.type().name()).data(ev.payload()));}catch(Exceptionignored){}}}

断线补拉靠Last-Event-ID：前端重连时带上次收到的最大 id，后端从回放 List 里把id > Last-Event-ID的事件补发一遍，再接 live 流。

④ 动态限流：注解式是全局桶，要按用户隔离得用 Registry 现取。

// 注解式 @RateLimiter(name="xx") 的 name 写死，全应用一个桶；// 要"每个会话一个独立桶"，用 RateLimiterRegistry 按 sessionId 现取：RateLimiterrl=registry.rateLimiter("rl:"+sessionId,"rl-default");returnrl.tryAcquirePermission();// 拿不到 = 被限流，返 429

四、实测输出：跨实例推送 + 限流 + 指标全到位

以下为本机真实运行（DeepSeek，实例 10004/10006，Redis 6391，Zipkin 9412）。

① 启动：FAQ 预载 12 条。

[FAQ缓存] 预载完成：从 docs/company-faq.md 入库 12 条 FAQ（key=lc4j-l13:faq:{归一化问题}，TTL=24h） Tomcat started on port 10004 (http) with context path '/' Started CustomerServiceApplication in 7.1 seconds

② 建会话发 token。

{"sessionId":"s-3a5c0d593bea4e77","token":"t-bfd4e34885e14149a3abdd834a295d89","expiresInSeconds":1800}

③ SSE 事件流——注意 id 乱序但单调（异步多线程所致）。

id:1 event:thinking data:{"stage":"receive","text":"收到问题，开始处理…","traceId":"6ac51756..."} id:4 event:thinking data:{"stage":"cache","text":"FAQ 缓存命中，零模型成本直接作答","traceId":"6ac51756..."} id:2 event:citations data:{"items":[{"source":"docs/company-faq.md",...}],"traceId":"6ac51756..."} id:3 event:token data:{"text":"公司实行弹性工作","traceId":"6ac51756..."} ... id:17 event:done data:{"traceId":"6ac51756..."}

看出来了吗？到达顺序是 1→4→2→3，但 id 严格单调。因为@Async是多线程池，紧挨着的事件被不同线程交错发出。前端要按 id 重排，别按到达顺序展示。

④ 限流：20 个并发打同一 session（10 次/秒的桶）。

Status codes: 200,429,200,200,200,200,429,200,200,429,429,429,200,429,429,429,429,200,200,429 429 count: 10

恰好 10 个 200、10 个 429——桶的容量就是 10。

⑤ 指标/actuator/metrics可查。

{"name":"cache.hits","measurements":[{"statistic":"COUNT","value":26.0}]}{"name":"ratelimit.rejected","measurements":[{"statistic":"COUNT","value":10.0}]}

⑥ 跨实例铁证——SSE 连 A(10004)，请求打到 B(10006)：

B(10006) HTTP 202: {"accepted":true}

结果：10004 的 SSE 连接收到了由 10006 产生的事件：

id:3 event:thinking data:{"stage":"receive","text":"收到问题，开始处理…","traceId":"6ac5195f..."} id:4 event:thinking data:{"stage":"model","text":"缓存未命中，走大模型路径…","traceId":"6ac5195f..."}

请求是 B 处理的，事件从 Redis 扇回了 A 的 SSE——横向扩展成立。

五、挑战题：改个参数，看看会怎样

1. ⭐调限流阈值：把按 sessionId 的桶从 10 次/秒改成 2 次/秒，重新并发打 10 次，观察 429 比例变化和ratelimit.rejected指标怎么涨。答案就在RateLimitService的配置里。
2. ⭐⭐演示断线补拉：问一个长问题，回答途中按 Ctrl+C 断 curl，重连时带上一次的Last-Event-ID，观察补发的历史帧。答案在SseEventBridge.replay()——它从回放 List 里把id > Last-Event-ID的事件补发。
3. ⭐⭐观察乱序：多问几个缓存命中题，注意 SSE 帧到达顺序和 id 不完全一致，想清楚为什么 id 单调但到达会乱（提示：@Async多线程池），以及前端该怎么展示。

六、生产环境进阶：三个加分项

1. Pub/Sub vs Redis Stream。Pub/Sub 是"发出去就不管"，离线订阅者丢消息（本课用回放 List 补这一刀）；Redis Stream 原生持久化 + 消费组 + ACK，是更强的方案。生产级要追不丢消息、要多消费组，直接上 Stream。

2. 缓存分层别过度工程。确定性 FAQ 用字面归一 Redis 缓存（省钱），模糊语义才上向量 RAG——"请假咋请"要命中"年假和请假怎么申请"才需要 embedding+KNN。别一上来就给所有问答上向量库。

3. 兜底不裸奔。任何故障路径（模型/工具/Redis）都要发 error 事件 + 人工客服话术，不能 500 白屏。熔断打开时前端显示"系统繁忙，已为您转接人工"。

七、面试回答模板

面试官：设计一个支持多实例、断线重连的 LLM 流式事件推送系统，怎么做？

一句话：事件统一信封（带 id/type/traceId），Spring 事件总线解耦业务和传输，Redis Pub/Sub 做跨实例扇出，回放 List 补 Last-Event-ID。展开：业务层只 publish AgentEvent，监听器用 Redis INCR 分配全局单调 id、convertAndSend 到每会话频道、RPUSH 留最近 100 条；各实例订阅频道，本地 Map 有这个 sessionId 的 SSE 才转发。浏览器连 A、请求打 B，事件经 Redis 扇回 A。（指向本课第一、三节 / lesson-13）

追问：限流为什么要按用户动态建桶？注解式限流不行吗？

注解式@RateLimiter(name="xx")的 name 写死，全应用一个桶，一个用户刷就把所有人的额度吃了。要按用户隔离，用RateLimiterRegistry按 sessionId 现取/现建独立桶。限流挡"太多请求"（入口），熔断挡"下游挂了"（出口快速失败走兜底），俩别混。（指向本课第三节）

追问：FAQ 确定性问答为什么这一课不用向量库？

FAQ 答案固定、问法收敛，把问题归一化后当 key 查 Redis，命中直接回标准答案，一次模型调用都不花（token 是钱）。向量检索解决"语义相似但字面不同"的模糊匹配，是分层不是替代——确定性命中走缓存，模糊语义才上 RAG。（指向本课第一、六节）

八、总结表

九、关于这个系列

本文是「Java 后端实战精通营」系列第 13 篇，原则：实战驱动、由浅到深、面试向，每篇文章的结论都可以亲手验证。

👉LangChain4j 实战精通营：https://gitee.com/j67mk2/langchain4j-journey

• 本文对应源码位置：lesson-13/（收官姊妹篇，核心看event/AgentEvent.java事件信封、event/AgentEventListener.javaRedis 网关出口、web/SseEventBridge.java本地转发与断线补拉、service/RateLimitService.java动态限流）

系列文章一览（主线 10 篇 + 生产增强 3 篇，按发布顺序）：

走到这里，这个系列就全部完结了。从第 1 课手写第一个 LLM 调用，到 RAG、工具调用、多轮记忆攒出企业客服（L10），再到生产增强——数据落中间件不怕重启（L11）、traceId 全链路可观测（L12）、事件总线 + Redis Pub/Sub 支持横向扩展与断线补拉（L13）。这条线的本质，是把一个"demo 级 LLM 调用"一步步练成"能上线的企业级智能体"：前面解决"AI 能做什么"，最后三课解决"怎么让它敢放到生产环境"。技术会过时，但"把 demo 做成系统"的这套拆解方法——解耦、降级、观测、横向扩展——是下一份工作天天要用的。

跑完有任何报错，把终端输出发评论区，一起排查。

标签建议：LangChain4j、Redis Pub/Sub、SSE
摘要建议（≤256 字）：单实例 SSE 能跑、多实例全废？本文把客服事件流重做成企业级方案：统一 AgentEvent 信封（带 id/traceId）、Spring 事件总线解耦业务与传输、Redis Pub/Sub 做跨实例扇出、回放 List 补 Last-Event-ID 断线重拉；再配 Resilience4j 按 sessionId 动态限流、模型重试熔断、Redis FAQ 缓存省 token、Micrometer 指标。双实例实测：SSE 连 A、请求打 B，事件照样流回浏览器；20 并发触发 10 个 429。附完整命令与面试模板，源码在 gitee lesson-13 可 clone 重跑。