LangChain4j 企业级收官事件总线 Redis Pub/Sub 断线补拉客服终于敢上线作者鱼宵 LangChain4j 实战精通营 · 第 13 篇上一课我们做出了逐字流式回答 思考面板 工具调用徽章的客服我本以为这下稳了。结果上线演练第一天就被运维泼冷水你这 SSE 事件是业务层直接拿SseEmitter发的单实例是爽多实例怎么搞浏览器连在 A 上、请求打到 BB 根本拿不到 A 上的 emitter事件往哪推我当时愣了一下。再一盘点这层自研事件链有四个死穴业务层和传输层耦死、事件没统一 schema、断线就丢事件、也没有限流重试和指标。说白了它就是个单店后厨直接端菜上桌的 demo——只在这一家店成立。这篇就把这层重做成企业级方案也是这个系列的收官篇。代码全在仓库的lesson-13/目录clone 下来照着命令起两个实例你能亲眼看到浏览器连 A、请求打 B事件照样流回浏览器。一、核心原理单店端菜为什么上不了连锁先认清 L12 事件链的四个死穴业务层和传输层耦死Service 里 new 了 SSE 对象想换 WebSocket、想加事件落盘、想多实例转发全得改业务代码。事件没 schemathinking 是个 Map、token 是裸字符串、done 是另一个 Map前端解析全靠猜。没有 id浏览器刷新/断网重连不知道从哪条补起直接丢一段回答。单实例绑死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序列化成 JSONconvertAndSend到每会话独立频道lc4j-l13:events:{sessionId}同时RPUSH进回放 List 留最近 100 条。每个实例都订阅这个频道收到后看本地 Map 里有没有这个 sessionId 的 emitter——有就转发给浏览器没有就丢。跨实例铁证浏览器 SSE 连在 A请求打到 B 产生事件B 发到 RedisRedis 扇出给所有实例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--filternameredis-l13--filternamezipkin-sa# 都是 Up 才继续java-version[bool][Environment]::GetEnvironmentVariable(DEEPSEEK_API_KEY)第 1 步编译打包。cd langchain4j-journey\lesson-13$env:JAVA_HOMEC:\Program Files\Java\jdk-17mvn clean install-DskipTests# BUILD SUCCESS第 2 步起实例 A前台跑能看日志。java-jar target\lesson-13-1.0.0.jar--server.port10004看到[FAQ缓存] 预载完成... 12 条 FAQ和Tomcat started on port 10004即成功。第 3 步浏览器开前端。http://localhost:10004/页面会自动建会话、连 SSE。第 4 步见证横向扩展再起实例 B。新开一个 PowerShelljava-jar target\lesson-13-1.0.0.jar--server.port10006浏览器 SSE 连在 10004但你用 curl 把请求打到 10006——事件照样在浏览器里流出来第四节有实测。三、关键代码事件信封 Redis 网关 本地转发① 统一事件信封 AgentEventid 不在业务层赋值交给 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 三处可对MapString,Objectpayload){publicenumType{thinking,tool,token,citations,done,error,rate_limited}// 业务层构造还没编号的事件id 由 Redis INCR 在监听器里分配跨实例单调A/B 不撞号publicstaticAgentEventof(Typetype,Stringsid,StringreqId,StringtraceId,MapString,Objectp){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.redisredis;this.jsonjson;}Async// 异步线程池跑业务 publish 完立刻继续不等 Redis、不等 SSEEventListenerpublicvoidon(AgentEventevent)throwsException{Longidredis.opsForValue().increment(lc4j-l13:seq:event.sessionId());// 发号Stringbodyjson.writeValueAsString(event.withId(idnull?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{privatefinalMapString,SseEmitteremittersnewConcurrentHashMap();// 本实例的 SSE 连接privatefinalStringRedisTemplateredis;privatefinalObjectMapperjson;publicSseEventBridge(StringRedisTemplateredis,ObjectMapperjson){this.redisredis;this.jsonjson;}OverridepublicvoidonMessage(Messagemessage,byte[]pattern){StringchannelnewString(message.getChannel(),StandardCharsets.UTF_8);StringbodynewString(message.getBody(),StandardCharsets.UTF_8);if(!channel.startsWith(AgentEventListener.CHANNEL_PREFIX))return;StringsessionIdchannel.substring(AgentEventListener.CHANNEL_PREFIX.length());SseEmitteremitteremitters.get(sessionId);// 本实例没这条 SSE → 丢掉if(emitternull)return;try{AgentEventevjson.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(namexx) 的 name 写死全应用一个桶// 要每个会话一个独立桶用 RateLimiterRegistry 按 sessionId 现取RateLimiterrlregistry.rateLimiter(rl:sessionId,rl-default);returnrl.tryAcquirePermission();// 拿不到 被限流返 429四、实测输出跨实例推送 限流 指标全到位以下为本机真实运行DeepSeek实例 10004/10006Redis 6391Zipkin 9412。① 启动FAQ 预载 12 条。[FAQ缓存] 预载完成从 docs/company-faq.md 入库 12 条 FAQkeylc4j-l13:faq:{归一化问题}TTL24h 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 个并发打同一 session10 次/秒的桶。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——横向扩展成立。五、挑战题改个参数看看会怎样⭐调限流阈值把按 sessionId 的桶从 10 次/秒改成 2 次/秒重新并发打 10 次观察 429 比例变化和ratelimit.rejected指标怎么涨。答案就在RateLimitService的配置里。⭐⭐演示断线补拉问一个长问题回答途中按 CtrlC 断 curl重连时带上一次的Last-Event-ID观察补发的历史帧。答案在SseEventBridge.replay()——它从回放 List 里把id Last-Event-ID的事件补发。⭐⭐观察乱序多问几个缓存命中题注意 SSE 帧到达顺序和 id 不完全一致想清楚为什么 id 单调但到达会乱提示Async多线程池以及前端该怎么展示。六、生产环境进阶三个加分项1. Pub/Sub vs Redis Stream。Pub/Sub 是发出去就不管离线订阅者丢消息本课用回放 List 补这一刀Redis Stream 原生持久化 消费组 ACK是更强的方案。生产级要追不丢消息、要多消费组直接上 Stream。2. 缓存分层别过度工程。确定性 FAQ 用字面归一 Redis 缓存省钱模糊语义才上向量 RAG——请假咋请要命中年假和请假怎么申请才需要 embeddingKNN。别一上来就给所有问答上向量库。3. 兜底不裸奔。任何故障路径模型/工具/Redis都要发 error 事件 人工客服话术不能 500 白屏。熔断打开时前端显示系统繁忙已为您转接人工。七、面试回答模板面试官设计一个支持多实例、断线重连的 LLM 流式事件推送系统怎么做一句话事件统一信封带 id/type/traceIdSpring 事件总线解耦业务和传输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(namexx)的 name 写死全应用一个桶一个用户刷就把所有人的额度吃了。要按用户隔离用RateLimiterRegistry按 sessionId 现取/现建独立桶。限流挡太多请求入口熔断挡下游挂了出口快速失败走兜底俩别混。指向本课第三节追问FAQ 确定性问答为什么这一课不用向量库FAQ 答案固定、问法收敛把问题归一化后当 key 查 Redis命中直接回标准答案一次模型调用都不花token 是钱。向量检索解决语义相似但字面不同的模糊匹配是分层不是替代——确定性命中走缓存模糊语义才上 RAG。指向本课第一、六节八、总结表坑现象解法业务层直接拿 SseEmitter单实例能跑多实例事件发不回去Spring 事件总线 Redis Pub/Sub 扇出事件没 id断线重连丢一段回答Redis INCR 分配 id Last-Event-ID 补拉事件没 schema前端解析全靠猜统一 AgentEvent recordid/type/traceId/payload注解式限流全局桶一个用户刷爆所有人额度RateLimiterRegistry按 sessionId 动态建桶SSE 帧乱序id 1→4→2→3 到达id 单调INCR 原子前端按 id 重排异步线程 MDC 丢 traceId事件 payload 里 traceId 为 null业务线程 publish 时抓 MDC 塞进信封Zipkin 属性名变了老spring.zipkin.base-url不生效Boot 3.x 用management.zipkin.tracing.endpoint事件管道故障拖死业务Redis 抖动把主流程打挂监听器 catch 住只打 warn监控是旁路九、关于这个系列本文是「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 篇按发布顺序篇主题1LangChain4j 第一个对话程序手写 LLM 调用链路看清模型怎么开口2LangChain4j 提示词工程模板 few-shot 温度提示词从写死到复用3LangChain4j 结构化输出让模型吐出 Java 对象JSON 不再手写解析4LangChain4j 对话记忆让模型记住聊过什么三种方式实测对比5LangChain4j 流式输出打字机效果 SSE回答不再干等三秒6LangChain4j 工具调用Function Calling 让模型动手查数据多工具连调实测7LangChain4j 向量检索文本怎么变成可搜索的坐标8LangChain4j RAG 流水线手写问答全流程给 AI 开卷考试9LangChain4j Agent多步推理自主调用放手让实习生自己安排工作10LangChain4j 企业智能客服RAG 工具 记忆 流式 兜底验收 5 项全过11LangChain4j 生产化改造记忆落 Redis、向量落 ES重启再也不丢12LangChain4j 可观测与透明化traceId 串起全链路SSE 直播思考过程13LangChain4j 企业级收官事件总线 Redis Pub/Sub 断线补拉客服终于敢上线走到这里这个系列就全部完结了。从第 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 重跑。