企业官网建设流程全解析

WMS（仓储管理系统）集成 AI Agent，前面七讲我们把智能体从"能聊天"一步步掰成"能干仓储活"：任务编排、工具调用、库存分析、异常决策、单据解析……讲真，到第八讲之前，我一直觉得架构和技术方案已经没什么可聊了。直到压测那个晚上，我坐在机房里盯着监控面板上的一堆数字发呆，才意识到生产部署和并发安全，才是前面所有设计能不能活下去的底线。这一篇作为终篇，我不聊业务、不聊模型，就聊一个坑：Semaphore 放进 Flux.defer，明明代码里写了并发限制，压测时却发现限流形同虚设，我们整个团队从晚上十点排查到第二天凌晨六点。

这个坑非常适合放在终篇来讲，因为它兼具"隐蔽性"和"杀伤力"。隐蔽在于，代码里很明确地写着Semaphore(10)，任何人扫一眼都觉得有并发控制；杀伤力在于，一旦限流失效，外部 AI 接口、数据库连接、MQ 消费者全都会被冲垮。WMS 这种每天承载几万单进出的系统，任何一个环节被打爆，都会引发仓库作业大面积停摆。这篇文章完整还原当时的压测方案、排查链路、根因分析和最终修复方案，如果你也在用 Spring WebFlux 集成 AI Agent，或者只是在自己的响应式服务里做了类似限流，我建议耐心看完，尤其是第 3 章和第 4 章。

1. WMS接入AI Agent后，系统的并发模型彻底变了

1.1 智能体在WMS里到底干什么，为什么非要限流

WMS 的系统特性，说白了就是"单据驱动、波次作业"。每天下单高峰、波次拣货、退货入库、盘点差异这些业务节点，会产生大量瞬时任务。以前这些任务走的是固定规则引擎，性能可控，QPS 预测得很准。接入 AI Agent 之后，情况完全变了：Agent 要在大模型、知识库、仓库数据、ERP 之间来回调度，单个任务的耗时从几十毫秒涨到几百毫秒甚至几秒，还非常吃外部 API 配额。

我们当时给 Agent 分配了几类核心工作：

• 智能上架推荐：根据库存周转率和库位约束给出上架策略
• 波次任务动态编排：遇到波次冲突时自动调整拣选路径
• 退货商品的处置决策：判断质检不合格商品走退货还是降价通道
• 异常单据推理：库存差异出现时，结合历史数据推断原因并给出处理建议

这些任务本身是异步提交的，但对同一家 AI 服务商而言，配额是死的。如果并发不受控，十几分钟内就能把 API Key 的配额打爆，接着 AI 服务开始返回 429，再有异常单据进来，就只能挂起等超时。所以"限流"不是锦上添花，是活下去的基本盘。

1.2 从同步到响应式：为什么选择了WebFlux

这里补一个背景，为什么是 Flux 而不是普通的 Runnable + CompletableFuture。我们最初把 Agent 调用包装成了同步方法，任务线程池一满，Tomcat 线程就全部阻塞在外部 API 调用上，线程数飙到几千，CPU 却闲着。换了 WebFlux 之后，整个链路变成非阻塞的：一个请求进来，线程发起调用就挂起，不占线程，IO 回来再恢复。这种模型对外部 API 调用多的场景很友好，吞吐量高得多。

但是非阻塞模型有个代价：几乎所有的并发控制组件，都默认你不是阻塞式的。Semaphore 这种经典并发工具，acquire()会阻塞当前线程，在 WebFlux 里如果你不小心让它阻塞了 Netty 的 Event Loop 线程，那就是新的灾难。这为后面的坑埋了伏笔。

1.3 Semaphore 的定位：给AI调用加一道闸

选型的时候我们对比过几套方案：

• Resilience4j 的 Bulkhead：功能全，但要引入整套框架，我们当时觉得有点重
• 手写 Semaphore：十几行代码搞定，简单直接，团队没人不会
• Redis 分布式限流：当时只有单实例部署，暂时不需要

于是选了 Semaphore。这也是当时所有问题的起点——不是 Semaphore 这个工具本身有问题，而是我们在响应式链里使用它的姿势不对。

2. 压测暴露的问题：三路并发全部失控

2.1 压测方案设计与预期值

先说压测方案。我们用 JMeter 做压力测试，场景是批量任务处理接口：POST /api/wms/tasks/process，每个批次带 10 个任务，服务端用 Flux 并行处理，预期受Semaphore(10)限制，任务内部调用 AI Agent 的平均耗时大概 800ms。

JMeter 的关键配置：

线程数：200 Ramp-Up：60 秒 循环次数：持续 20 分钟 HTTP 请求：POST /api/wms/tasks/process 定时器：高斯随机定时器，均值 500ms，偏差 100ms 监听器：聚合报告、响应时间图、TPS 曲线

启动压测前我们做了理论计算：Semaphore 许可证数量 10，单个 Agent 任务耗时 0.8s，那么在稳定状态下系统整体最大吞吐约10 / 0.8 ≈ 12.5 TPS。200 个并发线程压这个接口，肯定是大材小用，但我们关注的是"限流是否真的生效"，而不是"系统能跑多快"。

2.2 现场一：AI 接口被 429 淹没

压测开始后不到两分钟，AI 服务商的控制台就开始告警，我们自己的应用日志里也开始刷429 Too Many Requests。实时监控面板上，调用 AI Agent 的并发数在短短几秒内冲到了 150+，而我们的 Semaphore 申请的明明是 10。当时所有人的第一反应是"日志打错了？信号量被别的业务抢了？"，于是赶紧去查代码。

其实如果当时冷静下来想想，信号量是静态的、显式的，别的业务不可能抢走许可证。这个问题一定出在"我们使用 Semaphore 的方式"上。

2.3 现场二：数据库连接池耗尽，雪崩已经开始

AI 接口 429 只是第一个信号，紧接着是数据库。Agent 决策过程要查询库存、库位、商品基础信息，每完成一个决策就要写审计日志。HikariCP 的活跃连接数从平时的 20 左右一路涨到接近最大值，然后开始抛：

HikariPool-1 - Connection is not available, request timed out after 30000ms. java.sql.SQLTransientConnectionException: HikariPool-1 - Connection is not available

这一下连锁反应就出现了：不只 AI Agent 相关任务报错，正常的单据查询、上架、出库流程也开始超时——因为数据库连接全部被 Agent 任务占住不释放。当时的 RT 从平均 800ms 涨到 5 秒以上，错误率超过 15%。

2.4 现场三：CPU和内存都很闲，系统却被拖死

最诡异的是，应用服务器的 CPU 和内存指标都很好，CPU 才 20% 出头，内存平稳。典型的"系统没累，业务已断"。这个现象基本可以排除计算资源瓶颈，问题一定出在并发控制或下游依赖上。当晚我们就锁定了排查方向：为什么限流没生效。

3. 一夜排查：从JMeter曲线到Flux.defer源码

3.1 常规排查一批批被排除

先上 jstack 看看线程状态。抓完线程快照发现，没有大量线程阻塞在 Semaphore 上，大多阻塞在数据库连接获取和外部 HTTP 调用上。这其实已经是个重要线索：如果Semaphore(10)真的生效，应该有一堆线程在acquire()上排队，可事实是它们根本没在排队，直接就冲进后续调用了。

接着查配置。第一反应是"Semaphore 的许可证是不是配大了"？检查配置中心，permits=10，没错。第二反应是"是不是代码逻辑里许可证用到了别的地方"？全局搜索 Semaphore 引用，发现只在 Agent 调用链中出现了一次，没有其他持有者。

到这里，问题几乎被锁定在"Semaphore 实例本身"或"调用链的装配方式上"。

3.2 白天没注意的代码细节：Semaphore 在 Flux.defer 里

凌晨两点，有人开始在 IDE 里一行行读 Agent 调用链的代码。核心逻辑大概是这样的：

public Flux<AgentResult> executeBatch(List<Task> tasks) { return Flux.fromIterable(tasks) .flatMap(task -> buildAgentFlux(task), 64); } private Flux<AgentResult> buildAgentFlux(Task task) { return Flux.defer(() -> { Semaphore semaphore = new Semaphore(10); try { semaphore.acquire(); return callAiAgent(task) .doFinally(signal -> semaphore.release()); } catch (InterruptedException e) { return Flux.error(e); } }); }

这里一眼看上去"有限流、有释放、有异常处理"，当时代码 Review 还顺利通过了。但注意，new Semaphore(10)这行代码写在Flux.defer的 Lambda 里。在响应式链路里，这是一个埋雷的写法。

3.3 验证：写最小复现工程，现象立刻改变

为了严谨起见，我们没有直接下结论。先写了一个最小复现工程，验证"Semaphore 在 defer 里反复 new"的假设：

AtomicInteger invokeCount = new AtomicInteger(0); Flux.range(1, 100) .flatMap(i -> Flux.defer(() -> { Semaphore semaphore = new Semaphore(1); boolean acquired = semaphore.tryAcquire(); if (acquired) { invokeCount.incrementAndGet(); semaphore.release(); } return Flux.just(acquired); })) .blockLast(); System.out.println("invokeCount = " + invokeCount.get());

我们预期：Semaphore(1)限定后，invokeCount 最多等于 1。实际跑出来是 100。原因很简单：每次订阅Flux.defer时，Supplier 都被重新执行，Semaphore 被 new 了 100 次，每次都拿到新的 1 个许可证。看完输出结果，我们团队当时面面相觑——原来问题压根不出在"限流逻辑"，而出在"每个订阅者都自己建了一个闸门"。

3.4 根因确认：defer 的延迟执行 + 实例作用域

到这里，根因已经清楚了：Flux.defer 把数据源创建逻辑延迟到每次订阅时才执行，因此 Lambda 内部的"创建 Semaphore 实例"这个动作也被延迟且重复执行了。换句话说，只要存在多个订阅者，就存在多个互相独立的 Semaphore，每个许可证都是满的，整体限流完全失效。

把两个机制放一起看：

• Flux.defer 的语义：每次订阅时执行 Supplier，返回一个新 Publisher。
• Semaphore 的语义：同一实例内统计许可证数量，跨实例无法共享计量。

两者叠加，结果就是：Semaphore 的"全局性"被 defer 的"每次创建"彻底瓦解。我们当时以为做了一个"限流"，实际上只是一堆没人共享的计数器。

4. 为什么限流会失效：Flux.defer的订阅语义与Semaphore的作用域

4.1 Flux.defer 到底 defer 了什么

很多人对Flux.defer的理解停在"延迟执行"四个字，但到底延迟到哪里、重复几次，如果不做细致验证很容易误解。我再用一个通俗类比：

普通Flux.just(...)像你提前点了一道菜，厨师在你进门之前就做好了放那里等；Flux.defer像你每次下单时厨师才重新做，来一桌客人做一桌菜。所以：

• Flux.just在装配阶段（定义链路的时刻）就固定了数据；
• Flux.defer在订阅阶段（每次 subscribe 的时刻）才构造数据源。

在 WebFlux 里，每次 HTTP 请求进来、每次flatMap内部订阅子流，都可能触发一次subscribe。这也意味着defer的 Lambda 可能会被执行非常多次。

4.2 三种写法的对比：哪个限流有效，哪个形同虚设

我整理一张表，这是当晚排查结束后我们梳理出来的，贴在团队 Wiki 里供后来人参考：

第一种就是我们的坑：限流形同虚设。第三种是我们最终的修复方向。

4.3 为什么这个坑如此难抓

事后复盘，"抓了一晚上"有几个客观原因：

1. 代码审查的盲区：代码里确实有"Semaphore + acquire/release + 异常处理"的完整逻辑，常规 review 真不会专门检查"Semaphore 是不是单例"。
2. 字面误导：new Semaphore(10)里的数字 10 太显眼了，大家下意识以为"10 个许可证"是全局共享的。实际上在 defer 的庇护下，它变成了"每次执行都分配 10 个许可证"。
3. 缺失可观测性：我们没有给限流器加指标，比如"当前剩余许可证数""等待中的订阅者数"。如果上线前有这个仪表盘，压测后几分钟就能看出来available permits一直是 10，根本不用熬一宿。

这也引出一个更通用的判断标准：凡是包含"状态"的并发组件（Semaphore、CountDownLatch、AtomicInteger、线程池等），都要警惕它是否被放大到每次调用时重新创建。一个状态对象如果生命周期比"单次请求"还短，那它在并发控制上基本等于没干活。

4.4 顺便说清：什么时候应该用 Flux.defer

这次踩坑不能得出"Flux.defer 是坏东西"的结论。defer 有很多合理用途：

• 需要每次订阅时动态获取最新数据
• 想确保每次订阅都重新构造响应链，避免上游状态污染
• 和flatMap配合创建"每个订阅者独立"的一次性事件

真正要小心的不是 defer 本身，而是你把什么样的"对象创建"塞进了 defer。如果放进 defer 的是无状态的构造逻辑（比如构建 HTTP 请求参数、生成 UUID），没问题；如果放进 defer 的是有状态的并发组件，请三思。

5. 修复方案与压测验证：把Semaphore放回正确的位置

5.1 第一步：Semaphore 提升为类成员单例

修复的第一步很简单：把 Semaphore 实例从Flux.defer的 Lambda 内部提到类字段。这里已经不必须用 defer 了，直接构建响应链即可：

private final Semaphore agentLimiter = new Semaphore(10); public Flux<AgentResult> executeBatch(List<Task> tasks) { return Flux.fromIterable(tasks) .flatMap(task -> executeWithLimit(task), 64); } private Mono<AgentResult> executeWithLimit(Task task) { return Mono.fromCallable(() -> agentLimiter.tryAcquire()) .flatMap(acquired -> { if (!acquired) { return Mono.error(new RejectedExecutionException("Agent 并发已满，请稍后重试")); } return callAiAgent(task) .doFinally(signal -> agentLimiter.release()); }); }

现在 Semaphore 是类成员变量，整个应用只有这一个实例，所有订阅者共享这 10 个许可证。限流才真正生效。

5.2 第二步：用 tryAcquire 替代阻塞式 acquire

这一步很关键。原来代码里用的是acquire()，这个方法是阻塞的。在 WebFlux 的非阻塞线程模型里，一旦acquire()让 Netty Event Loop 线程挂起等待信号量，整个应用的事件循环就会卡住，吞吐量会断崖式下跌。就算限流有效，这种阻塞方式也会让系统"活得很憋屈"。

改成tryAcquire()之后，拿不到许可证立即返回 false，走快速失败逻辑，不会阻塞线程。WMS 这种场景下，任务提交方本来就是异步的，拿不到许可证直接返回"稍后重试"是完全合理的，并不会丢任务。

5.3 第三步：doFinally 保证许可证一定归还

这里还有一个容错细节：信号量的许可证释放必须覆盖所有结束路径。响应式流中有三条结束路径：正常完成（onComplete）、异常终止（onError）、中途取消（cancel）。如果只用doOnTerminate，它只能覆盖前两种，用户或下游取消订阅时不会触发，许可证就会泄漏。我们用doFinally来解决，它在任何结束信号（包括 cancel）时都会执行回调：

return callAiAgent(task) .doFinally(signal -> agentLimiter.release());

这个细节很重要。如果团队里有人把doFinally换成doOnTerminate，压测时间一长，Semaphore 的可用许可证会逐步归零，限流又从"宽松"走向"锁死"，同样非常麻烦。

5.4 修复后的压测数据对比

修复后重新压测，同一份 JMeter 脚本、同一台压测机，数据对比：

数据恢复到了理论预期（10 个许可证、单任务 800ms，TPS 大约 12）。最直接的变化是，AI 服务商那边不再告警，HikariCP 也不再超时。那一晚上消耗的精力，本质上就为了看懂一行new Semaphore(10)放错了位置。

修复之后，我们顺手补充了限流器的可观测性指标，包括当前剩余许可证数、拒绝请求数、信号量等待时长，把这些接到 Prometheus + Grafana。这让我觉得安心多了，至少下次再出问题，不用再靠肉眼看日志。

6. 终篇收尾：WMS集成AI Agent的生产部署建议

6.1 并发安全之外，部署AI Agent还踩过的几个坑

Semaphore 那个坑是最大的一颗雷，但不是唯一的。趁着终篇，把生产部署中其他值得讲的问题一起列出来：

1. 外部 AI 调用必须配超时。我们一开始没给 WebClient 配 connectTimeout 和 readTimeout，某次 AI 服务端挂起，Agent 任务全挂在网络等待上，把信号量许可证全部占满，后续任务全部快速失败。后来统一配置了connectTimeout=3s、readTimeout=30s。
2. Agent 的流式响应不要盲目拼接。某个任务用流式接口接收大模型回复，有人在最终collectList时没做大小限制，一次异常回复把内存撑爆。建议对流式内容做长度上限和背压控制。
3. 多实例部署时本地 Semaphore 只能管住自己那台机器。我们后来从单实例扩到两台，每台Semaphore(10)，对外总并发就成了 20。如果严格需要全局统一限流，要用 Redis 或网关层限流。
4. 不要把 Agent 的可重试逻辑做得太激进。WMS 任务有请求幂等，重试 2 次没问题，但重试如果带着信号量同步重试，等于一次失败占用多倍许可证时间。

6.2 可观测性建设：让下次排查不用再熬一宿

这次事故给的最大教训不是"Semaphore 要放对位置"，而是**"并发组件必须要能被看出来在没在干活"**。我们事后给所有限流器都加了 Metrics：

• available_permits：当前可用许可证
• waiting_threads：等待许可证的请求/订阅者数
• rejected_total：被拒绝的请求总数
• acquire_duration：获取许可证的耗时分布

有了这些指标，像"Semaphore 形同虚设"这种问题，压测开始后 5 分钟就能从 Grafana 上看出端倪——available_permits一直是满的，没有一个请求被拒绝，说明限流器根本拦不住东西。不用等到 AI 接口 429、数据库连接超时全都爆发出来，再顺着日志一层层往回摸。

6.3 给后来者的三点建议

最后说三点实在的，都是拿这一晚上的血泪换的：

• 看到响应式链里的"并发组件"，先问一句"它是单例吗"，再问一句"它会不会阻塞 Event Loop"。这两个问题能过滤掉八成以上的并发问题。
• 压测不是上线前的仪式，是给系统找出"假装在控制并发"的地方。每次压测至少要看限流器的指标，而不是只看 TPS 和错误率。
• 写代码时，把"有状态组件"往方法外部放，把"无状态逻辑"留在 defer 或函数式链路内部。边界划清楚，踩坑概率会低很多。

WMS 集成 AI Agent 的这八讲到这里就结束了。从架构设计到生产部署，技术栈在这些文章里是次要的，真正有价值的是这些用熬夜换来的"为什么"。如果这篇终篇能帮你少熬一个通宵，那一晚上就值了。