企业官网建设流程全解析

WMS仓储系统集成AI Agent这套项目一路走到最后一讲，最意外的坑不在模型效果，反而出现在生产部署时的并发安全：Semaphore放进Flux.defer的写法，让整个服务在压测并发数上来后直接雪崩。我们用Jmeter抓了一个晚上，先后排除了网络、超时、AI服务端限流，最终靠线程Dump锁定了问题根源——信号量获取操作阻塞了Netty事件循环线程。这一篇就把这段完整经历写下来，包括最初的设计误区、通宵排查的思路、修复后的正确写法，以及几个压测和并发控制的通用经验。如果你也在用Spring WebFlux或Project Reactor对接外部AI接口，并且打算用Semaphore控制并发，这篇文章能帮你少走一晚上的弯路。

1. 场景还原：WMS 为什么需要给 AI Agent 套并发闸门

1.1 从仓储作业到 AI 流式交互

我们做的不是简单的“聊天机器人”，而是把AI Agent嵌进仓储系统的作业链路。比如操作员在PDA上问“A区还有多少库位能放中型件”，Agent接到问题后要拆解语义、查询库存、结合库位占用情况生成回答；再比如波次释放时，Agent需要根据当前拣货任务密度给出分批建议。这些交互涉及多个内部服务调用，并且最终要对接大模型服务商的流式接口。

集成方式上，我们选择了WebFlux。原因很朴素：AI接口的响应是流式的，WebFlux的Flux天生适合做流式转发；另外WMS虽然是传统业务系统，但仓储现场的移动端操作频率不低，我们希望网关层能扛住较高的并发连接数，而不是为每个请求占用一个线程。这种背景下，Flux.defer在项目里被大量使用，初衷是保证每次订阅能拿到最新的上下文。

但随之而来的问题也很直接：大模型服务商给了并发限制，按账号维度只能同时处理有限个请求。一旦我们把AI能力开放给整个仓储端，现场几十号人同时操作，瞬间就能把服务商的并发额度打满。超出的请求要么被对方限流返回429，要么因为排队时间过长拖垮用户体验。更麻烦的是，我们还要保证仓库核心的库存查询、任务下发不被AI请求抢走资源。

所以在接入层必须做一道闸门，控制同时调用AI接口的请求数量。这道闸门既要能快速放行，又必须在并发达到上限时让多余的请求等待，而不是无脑拒绝。

1.2 限流方案对比：我们为什么选了 Semaphore

当时捋了一遍可用的方案，列在下面：

• 分布式限流组件（Redis + Lua）：适合多实例部署，可以做全局限流，但需要维护Redis脚本，且有一个常见的坑：并发请求同时抢锁时，瞬时流量容易把Redis连接池打满。
• Resilience4j RateLimiter：基于令牌或滑动窗口，语义和信号量不同，我们希望控制的是“同时在飞”的流式请求数，而不是“每秒能放行多少次”。
• Hystrix信号量隔离：可以控制并发，但重，而且Hystrix已经进入维护模式，引入它只为了一个信号量功能不划算。
• JUC Semaphore：JVM内置，非公平/公平模式可选，性能足够，语义正好是“允许多少个线程同时通过”。

我们的AI服务网关是单实例部署（WMS系统本身有多个实例，但AI网关单独拆了一层，暂未做多活），所以不需要跨进程限流，Semaphore够用，也最简单。另外，流式接口的特点是连接建立后要保持一段时间，这段连接占用的不是线程而是连接数和后续的AI服务端配额，用令牌桶按“每秒”限制并不贴合，我们需要的是“当前同时处理多少个”，信号量是最贴合的模型。

现在回头看，选型没有错，错的只是把它用在了不合适的位置。

2. 第一版实现：把 Semaphore 塞进 Flux.defer

2.1 第一版代码长什么样

先看最初提交的代码，问题就藏在里面：

@Component public class AiAgentGateway { private static final Logger log = LoggerFactory.getLogger(AiAgentGateway.class); private final AiAgentClient aiAgentClient; // 全局共享信号量，允许同时处理 20 个 AI 请求 private final Semaphore aiSemaphore = new Semaphore(20); public Flux<String> chat(String userQuery, String warehouseId) { return Flux.defer(() -> { try { // 尝试获取许可，获取不到就阻塞等待 aiSemaphore.acquire(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return Flux.error(e); } return aiAgentClient.streamChat(userQuery, warehouseId) .doFinally(signalType -> { // 流结束或取消时释放许可 aiSemaphore.release(); log.info("AI signal released, type={}", signalType); }); }); } }

当时写这段代码的思路看起来“没毛病”：Flux.defer能确保每次订阅时才执行acquire，避免服务启动时或Flux装配时提前占用信号量；doFinally能在流正常结束、异常结束、甚至客户端断开时确保release；日志还能打印释放时机方便排查。我们甚至在单元测试里模拟了20个并发订阅，逻辑也全对，信号量能正常释放。

问题是，单元测试里我们用的是普通的线程池，而真实运行环境中，订阅这个Flux的线程是Netty的Reactornetty线程。这一步之差，成了压测事故的引爆点。

2.2 藏在这段代码里的三个雷

第一颗雷：阻塞调用发生在事件循环线程上。WebFlux处理HTTP请求时，最终订阅响应式流的线程通常就是Netty事件循环线程（reactor-http-nio-x）。事件循环线程的本质是循环处理注册到它上面的所有Channel的IO事件，它不能被阻塞，一阻塞就是全局性灾难。Semaphore.acquire()在没有可用许可时不释放CPU，而是让当前线程进入WAITING状态，这正好把事件循环线程钉死。

第二颗雷：信号量释放的时序没考虑清楚。doFinally确实能在终止信号时触发，但它触发的前提是下游订阅者已经准备好接收信号。如果在流尚未建立时，上游构造阶段就抛异常，或者aiAgentClient内部用延迟订阅的方式创建流，那doFinally不一定能立刻释放。一旦release少了一次，信号量的许可数就永久少了一个，积累到一定程度后，所有请求都会卡在acquire上。

第三颗雷：也是最讽刺的，我们把Semaphore放进Flux.defer的supplier里，看起来是把“获取许可”延迟到了订阅时，但并发安全最怕的就是“静态的语义”落在“动态的线程上下文”里。这段代码没有任何synchronized或原子性问题，Semaphore本身也是线程安全的，但它让阻塞行为污染了Reactor的调度模型，已经超出了“并发安全”本身的范畴，变成了“线程饥饿型故障”。

白天试了几个请求一切正常，因为并发量不超过20，acquire直接返回，根本不会阻塞。压测一上并发，问题才暴露，这也是为什么我们抓了一整个晚上才反应过来。

3. 压测抓了一整晚：现象、排查与根因

3.1 Jmeter 压测步骤与第一轮数据

先说我用Jmeter压测时的配置步骤，方便还没踩过坑的同学直接参考：

1. 打开Jmeter，创建线程组：Thread Group。压测AI流式接口，线程数设为100，Ramp-Up时间10秒，循环次数5。
2. 添加HTTP请求：Sampler选HTTP Request，协议填http，服务器写网关地址，路径写/ai/chat，Method选POST，Body Data填JSON参数。
3. 添加聚合报告：Listener选择“聚合报告”，运行结束后能直接看到平均响应时间、TPS、错误率。
4. 添加结果树：用于看单个请求响应内容，排查返回值是否正常。
5. 运行测试，在线程组里勾选“调度器”，设置持续时间60秒，固定压测时长。

第一轮压测数据就很诡异：100个并发线程，20个能正常拿到流式响应，剩下80个请求全部超时，超时时间我设的是30秒。TPS峰值只有20左右，然后一路掉到了0。最要命的是，AI接口挂掉的同时，网关服务对其他接口的响应也跟着变慢，连/actuator/health健康检查都开始出现偶发超时。这时候基本可以判断不是AI服务商的问题，而是我们自己的网关服务整个被拖住了。

接着我用命令抓了几次线程快照：

jstack -l <pid> > threaddump1.txt # 等10秒再抓一次，便于对比线程状态变化 sleep 10 jstack -l <pid> > threaddump2.txt

打开threaddump文件，grep出Netty线程组：

"reactor-http-nio-1" #16 prio=5 os_prio=0 cpu=0.20ms elapsed=123.45s tid=0x... nid=0x1a2b waiting on condition [0x...] java.lang.Thread.State: WAITING (parking) at jdk.internal.misc.Unsafe.park(Native Method) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:194) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(...) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(...) at java.util.concurrent.Semaphore.acquire(Semaphore.java:... at com.xxx.wms.ai.AiAgentGateway.lambda$chat$0(...) at reactor.core.publisher.FluxDefer.subscribe(...)

连续看到多个reactor-http-nio-x线程都停在Semaphore.acquire()上，那一刻真相已经很明显了：Netty的事件循环线程几乎全部阻塞在信号量获取上，没阻塞的线程也因为其他Channel等待IO事件积压而接近瘫痪。

3.2 线程 Dump 抓到了什么

线程Dump是这次排查的决定性工具。除了看到大量WAITING (parking)状态的Netty线程，我还注意到几个细节：

• 阻塞点统一是reactor-http-nio-x，说明订阅确实发生在事件循环线程上，而不是我们以为的业务线程池。
• Semaphore.acquire进入的是AbstractQueuedSynchronizer的队列，等待线程数量远超许可数，意味着大量请求不仅在排队，还把事件循环线程都押成了“人质”。
• 有个线程栈显示当前正在处理WebFlux路由分发，中间夹着自定义Filter，说明这个线程是从HTTP请求链路一路走到订阅的。

这时候我才真正意识到问题的严重性：信号量本来就是用来让一部分线程等待的，但等待的这些线程是Netty的IO线程，它们管着成千上万个连接的心跳、读请求、写响应。它们一停，服务表面上进程还活着，其实整个网关已经“假死”。

3.3 根因链条：Flux.defer、订阅线程与阻塞 API

把完整根因链条补充清楚，方便以后排查类似问题。

Flux.defer的作用是延迟数据源的创建：调用Flux.defer(supplier)的时候，supplier不会立即执行，而是等到有人subscribe时才执行，并且每次subscribe都会重新执行一次。这个特性本身没问题，问题在于谁触发了subscribe。

在WebFlux接口方法中返回Flux时，Spring WebFlux框架会在Netty事件循环线程上对返回值进行订阅和调度。此时Flux.defer中的supplier就在这个线程上同步执行。我们把Semaphore.acquire()放在supplier里，等于让事件循环线程去抢信号量，抢不到就park在那里不动。

这个链条的出事条件是：并发订阅数 > 信号量许可数。只要触发，就不是“部分请求变慢”这种轻柔故障，而是把基础IO线程池耗尽，进而拖垮整个服务。

另外，如果当时我们把Semaphore声明成局部变量，每次调用都新建一个，那就是另一个事故现场：所有请求都会放行，并发控制完全失效，AI服务商的接口会直接被打爆。我们虽然没有犯这个错，但排查时一度怀疑过，也作为候选方向验证过，所以一并写在文章里。

4. 修复方案：反应式并发控制可以这样写

4.1 方案一：把阻塞获取丢给弹性线程池

方向其实很简单：信号量还要用，但绝对不能阻塞Netty事件循环线程。正确的做法是，在独立调度器上获取信号量，拿到许可后再切换到AI客户端流式调用。

改造后的核心代码：

@Component public class AiAgentGateway { private static final Logger log = LoggerFactory.getLogger(AiAgentGateway.class); private final AiAgentClient aiAgentClient; private final Semaphore aiSemaphore = new Semaphore(20); // 专门用于阻塞操作的线程池 private final Scheduler semaphoreScheduler = Schedulers.newBoundedElastic( 100, // 线程池上限 1000, // 等待队列上限 "semaphore-pool" // 线程名前缀 ); public Flux<String> chat(String userQuery, String warehouseId) { return Mono.fromCallable(() -> { // 阻塞获取信号量，但发生在独立线程池中 aiSemaphore.acquire(); return true; }) .subscribeOn(semaphoreScheduler) .flatMapMany(ignore -> aiAgentClient.streamChat(userQuery, warehouseId) .doFinally(signalType -> aiSemaphore.release() ) ); } }

这段代码的关键点有两个：Mono.fromCallable负责把“获取信号量”变成一个异步任务；subscribeOn(semaphoreScheduler)让这个任务在独立线程池中执行，而不是Netty事件循环线程。拿到信号量之后，flatMapMany才真正调用AI客户端，流式的数据返回路径仍走Reactor自己的网络线程，不会因为阻塞API拖累主线程池。

注意，这里有个容易踩的坑：不能用publishOn代替subscribeOn。subscribeOn影响的是“从信号量获取到flatMapMany创建”这一段的上游执行线程；publishOn影响的是下游线程，两者语义完全不同。我第一次改的时候就是写成了publishOn，结果阻塞照样发生在事件循环线程上，压测依然炸。

Schedulers.newBoundedElastic里我设置了队列上限1000，避免突发流量把线程池和队列都打满导致内存溢出。信号量本身有20个许可，真正会在线程池里等待的请求，最多也就是1000个在排队，后面更多请求会直接走拒绝策略。

4.2 方案二：用 Reactor 背压替代信号量

如果不想引入Semaphore，也可以利用Reactor自带的并发控制机制，最常用的是limitRate和concatMap。不过它们和信号量的语义不完全一样。

limitRate(20)表示每秒最多向下游发送20个元素，更适合做吞吐控制。我们的场景是流式响应，一个请求对应一整个Flux，limitRate限制的是单个Flux内的元素速率，限制不了“同时存在的Flux数量”。

concatMap配合自定义调度器可以控制并发订阅数量，但写起来繁琐，而且对上游流的订阅时序要求更严格。Flux.merge可以设置concurrency参数，但它控制的是同时订阅的内部流数量，同样不是并发连接数。

实际上，还有一种更贴合“信号量隔离”的方案：自定义Scheduler，用固定大小的线程池执行AI调用。但AI客户端本身是异步的，不占用线程，所以线程池大小限制不住并发数量，最终还是要靠信号量或类似的计数器来限流。

所以方案二更适合的场景是：如果你控制的是“每秒请求数”而不是“同时连接数”，用RateLimiter或Reactor的limitRequest更自然。像我们这种流式长连接场景，信号量依然是更准确的工具。

4.3 最终代码落地与压测验证

最终我把方案一落地，又额外加了几个细节：

• 信号量采用公平模式：new Semaphore(20, true)。公平模式下，等待已久的请求会优先获得许可，避免某些请求过度饥饿。代价是吞吐略低，在20并发规模下完全值得。
• 获取许可前先判断当前队列积压：如果semaphoreScheduler的队列已满，直接用Flux.error返回一个“系统繁忙”的提示，避免无限等待。
• 在doFinally里多加了一层防御：如果重复释放，会抛出IllegalStateException，我在日志里记录了release的次数和信号量剩余许可，方便监控。
• 添加指标上报：每次acquire/release后记录当前许可证数量，接入Prometheus的Gauge指标。

压测配置不变，还是Jmeter的100并发线程、60秒持续时间，结果对比很明显：

一个细节值得提：压测期间AI流式接口偶尔会有慢请求，最长的一次跑了8秒才返回完整内容。修复前这种慢请求会占用信号量20秒以上，导致其他请求长时间排队；修复后由于acquire发生在独立线程池，即使某个AI流处理慢，Netty事件循环也能继续处理其他连接，整体服务不会陷入瘫痪。

5. 终篇想说的几句肺腑之言

这次排障抓了一晚上，最有价值的收获不是改了几行代码，而是彻底看懂了一件事：并发安全在响应式编程里是一个“上下文敏感”的问题。同一个Semaphore.acquire()，放在普通业务线程里只是让这个线程等待，放在Netty事件循环线程里就是拖垮整个服务的导火索。写代码时不能只看“线程是不是安全的”，还要看“这段逻辑会在哪个线程上执行、会阻塞谁”。

几个实际的建议：

• 生产环境部署前，压测一定要做“超并发”场景，不能只测正常负载。我们白天人工测试一切正常，就是因为并发数根本没碰到信号量上限。
• Jmeter线程组不要只测几百个请求，要配合持续时长压测，比如至少跑3到5分钟，观察TPS曲线是否平稳，千万不要跑完一轮就说没问题。
• 遇到服务假死，第一反应应该看线程Dump，而不是反复重启服务。线程Dump能直接告诉你哪些线程在等什么锁、谁占着资源不放。
• doFinally确实能保证信号量释放，但它不是万能的。对于异步客户端和延迟订阅的流，一定要在真实压测环境验证释放次数和信号量泄漏情况。

如果你也正在做AI Agent类的网关集成，我建议提前把这些并发边界想清楚：信号量放哪个线程池、释放时机的兜底策略是什么、流式请求的取消和超时怎么和release配合。别看这些细节小，生产环境里一个信号量泄漏都可能造成整个服务“假死”，那就不只是加一个晚上的班能解决的问题了。