前阵子我们在给一套 WMS 仓储系统接 AI Agent 能力，做到第八讲也就是终篇的时候，差点被一个 Semaphore 的坑拖死在压测现场。当时整个链路已经是 Spring WebFlux 的响应式架构，AI 能力封装成异步流，接口层用了 Flux.defer 做订阅隔离，逻辑上看着没什么问题，结果并发一上来，系统直接雪崩。抓了一晚上线程栈和信号量日志，最后定位到的根因说出来你可能不信——Semaphore 放错了位置，放在 Flux.defer 里面，等于每个请求都拿到了一个全新的信号量，限流形同虚设。这篇把整个排查过程、底层原理和最终的生产级写法完整复盘一遍，给正在做类似集成或者准备上压测的兄弟一点参考。

WMS 这个场景有个特点：它不像互联网 C 端应用那样流量模型单一，出入库、盘点、波次调度、库位分配每时每刻都在产生任务，而且这些任务往往要实时调 AI 去做决策，比如库存异常检测、拣货路径优化、库容预测。生产环境里并发来源很杂，有 WCS 设备回报、RF 枪扫描、B 端商家 API 的突发流量，甚至有凌晨批量补数据的定时任务。AI 推理接口本身又不是弹性的，模型服务能扛住的并发是固定的，超了就只能排队或者直接超时。所以必须在上游加一道并发闸口，控制同时进入 AI 服务的请求数。我们第一版方案选的是 Semaphore 配合 Reactor 的 Flux.defer 来做。

—— 正文开始 ——

1. 为什么 WMS 集成 AI Agent 必须做并发控制

1.1 响应式架构下的限流需求是从业务场景长出来的

先交代一下背景。这套 WMS 是典型的仓储中台，底层数据模型有库位、库存、单据、批次、容器这些核心实体，上层业务则围绕收货、上架、拣选、复核、发货展开。集成 AI Agent 之后，系统里多了一层"智能决策引擎"，它会根据实时库存和单据数据输出建议，比如自动分配库位、推荐拣货顺序、预判爆仓风险、生成补货策略。

问题在于，这些 AI 决策并不是纯离线计算的。有一部分是同步请求，比如 RF 枪扫到箱码之后，系统要在几百毫秒内返回"该放到哪个库位"；还有一部分是异步批处理，比如夜里把几万个 SKU 的动销数据送到模型里做库容预测。同步请求天然会受 AI 服务的响应时间约束，异步任务则会在突发时刻把并发直接拉满。

生产环境里真正的并发来源我们摸过底，大概三类：

• WCS 立体库设备在高峰时段每分钟回报上千条任务状态，触发实时库位重算；
• 商家侧 ERP 对接的 API 在整点批量同步单据，瞬时 QPS 能冲到几十；
• RF 手持终端在波次拣货时集中扫码，每个动作都可能触发一次 AI 调用。

如果不对这些流量做闸口限制，AI 模型一挂，整个 WMS 的作业面都会停摆，库内作业员拿着 RF 枪干等，比接口报错更难受。所以并发控制这件事不是架构洁癖，是仓储业务持续运转的硬要求。

1.2 为什么选 Semaphore 而不是其他限流方案

当时团队里有人提议用 Redis 做分布式限流，也有人提议用 Resilience4j 的 RateLimiter，还有人说直接在网关层限流。我逐一分析过，最后选了 JVM 内的 Semaphore，理由其实挺务实的。

第一，AI Agent 的调用是单体应用内部的进程内调用，并不需要跨节点共享限流状态。我们只有两台应用节点，每台节点自己限制进入 AI 服务的并发数，加起来就是总并发上限，完全可以建模。引入 Redis 限流等于为了一个简单的并发控制引入分布式一致性负担，还得处理 Redis 本身抖动的问题。

第二，Semaphore 的语义和我们的需求几乎完全匹配。RateLimiter 限的是"每秒多少个请求"，Semaphore 限的是"同时有多少个请求在处理"，我们要的是后者。AI 推理接口的瓶颈是并发连接数而不是 QPS，同一时刻进去的请求越少，模型服务的推理延迟就越稳定。

第三，响应式链路上用 Semaphore 很顺手，可以配合 flatMap 操作符把许可的获取和释放封装成异步流程，跟 Reactor 的线程模型融合得比较好。如果用 synchronized 或者 Lock 来做，在响应式线程上会 block 住 event loop 线程，那才叫灾难。

不过这个选择也埋了一个大坑——Semaphore 本身没问题，问题出在它的实例被放在了 Flux.defer 的工厂方法里创建。当时代码长这样：

public Flux<String> invokeAgent(AgentRequest request) { return Flux.defer(() -> { Semaphore semaphore = new Semaphore(10); return Mono.fromCallable(() -> agentService.predict(request)) .doFirst(() -> semaphore.acquire()) .doFinally(signal -> semaphore.release()); }); }

看过这段代码你可能已经发现问题了。没错，每个订阅者执行到 Flux.defer 的 Supplier 时，都会 new 一个全新的 Semaphore，每个信号量都有 10 个许可。并发请求一多，实际同时进入 AI 服务的请求数就不是 10，而是 10 乘以订阅者的数量。这个坑不压测到一定量级根本不会暴露，因为低并发时每个请求都能轻松拿到自己的许可，表现得相当正常。

2. Flux.defer 打开的黑盒：Semaphore 为什么会在里面"分身"

2.1 先搞懂 Flux.defer 和普通 Flux 创建方式的区别

要理解这个坑，得先弄明白 Flux.defer 到底做了什么。Reactor 里的 Flux 分两种：冷发布者和热发布者。Flux.defer 创建的属于"冷流"里比较特殊的一种——它不在声明时生成数据序列，而是延迟到每次订阅时才执行工厂方法，重新生成一份 Flux 实例。

用生活里的例子打比方，Flux.just(1, 2, 3) 像是拍好的一部电影拷贝，无论多少个人来看，放的都是同一份内容；Flux.defer(() -> Flux.just(generate())) 则像是电影院现场售票、现场放映，每位观众进场的瞬间才决定片源是什么。

这个机制本来是为"每个订阅者都应有独立数据流"这种语义准备的。比如我们系统里根据当前库存快照生成波次任务时，希望每次订阅时重新读取数据库，避免多个订阅者共享同一个过期的快照，这时候用 defer 就非常合适。但如果你在 defer 工厂里创建了一个资源对象，比如 Semaphore，那么每次订阅都会创建一个新的资源。这跟"每次订阅都重新读取数据"的初衷就对不上了——资源是要跨订阅共享的，你却在每个订阅里各造一份。

2.2 信号量从"全局闸门"退化成"一次性门禁"

正常设计 Semaphore 的思路，是整个系统只有一个闸门，所有请求在门口排队，一次放行 N 个进去办事。Semaphore 放进 Flux.defer 之后，相当于每个请求进来时都现场砌了一个闸门，闸门上写着"限流 10 人"，然后请求自己通过这个全新闸门。所有请求都在自己的闸门里畅通无阻，系统的全局并发控制彻底失效。

我后来写了个小实验验证这个推理，用日志打印信号量对象的 hashCode。预期是每个请求拿到的 Semaphore 应该是同一个对象，最少 hashCode 要一致。实际跑出来的结果是，每来一个请求，hashCode 都变。这就有实锤了：Semaphore 的实例化发生在每个订阅者的执行路径上，而不是全局唯一的对象。

除了 Semaphore 之外，其他带状态的对象放进 defer 也会踩类似的坑，比如线程池、数据库连接池、定时任务调度器、限流令牌桶，甚至持有一个计数器的对象。只要这个对象需要跨订阅共享，就不能放在 defer 工厂里创建。这是响应式编程里一个很容易被忽略的边界条件。

2.3 为什么压测初期风平浪静，跑到深夜才炸锅

最开始我们在测试环境只做了功能验证，并发压到 5、10、20，每一档都跑得挺稳。从现象上看，限流确实派上了用场——每个请求都能正常拿到许可，AI 接口的延迟也控制在 300 毫秒以内。所有人都以为这个方案是没问题的。

其实这个阶段用的是温水煮青蛙的方式，问题在累积。每个请求都会 new 一个 Semaphore，也就是创建 10 个许可，虽然请求进来了，但它们实际上都在并发执行。低并发时，模型服务勉强能扛住，数据库连接池和下游 ERP 接口也还没到极限，所以表现正常。等我们把并发提到 50、跑持续压测半小时以上，AI 服务开始频繁超时，线程等待数暴涨，整个系统的响应时间从 200 毫秒一路飙升到十几秒。

这里有个通用教训：只做小规模压测暴露不了并发安全问题，要破坏性压测、持续压测，才能看到资源的真实竞争关系。你看着信号量在工作，其实它从来没真正拦住过流量，只是挡住了所有安全的人类直觉。

—— 正文继续 ——

3. 一晚上压测实录：从"以为下游慢"到"抓到真正元凶"

3.1 Jmeter 压测怎么配置才能"压得准"

先说压测工具。我们用 JMeter 做持续压测，配置其实不复杂，但有几个细节直接影响结论质量。按下述方式配，基本能压出生产问题的真实面貌。

第一步，新建线程组，设置线程数为 50，Ramp-up 时间为 5 秒，循环次数设置为"无限"，并勾选调度器，持续运行 30 分钟。Ramp-up 太短会导致瞬间流量冲击，把问题放大为"系统崩溃"而掩盖了限流失效的真相；Ramp-up 太长又达不到并发高峰的测试目标。5 秒内把 50 个线程全部拉起，是比较合理的渐进式加压方式。

第二步，添加 HTTP 请求取样器，配置我们内部 AI Agent 网关接口的路径和参数。注意这里参数最好用一个 CSV 数据集配置，每个线程用自己的请求数据，避免所有线程拿着同一份参数压同一个缓存 key。当时我们第一批压测脚本偷懒用了固定参数，结果命中缓存之后响应时间全是 1 毫秒，什么问题都测不出来。

第三步，添加聚合报告和查看结果树两个监听器，另外强烈建议加一个"jp@gc - Active Threads Over Time"插件，实时观察线程活跃度的曲线。真正有参考价值的不是平均响应时间，而是 90% 线、99% 线和错误率随时间的变化曲线。压测不是跑完看总量，是看过程。

第四步，压测期间盯三张图：活跃线程数变化、响应时间分布、JVM 堆内存使用。我们就是靠这三张图，把问题定性为"限流失效导致下游过载"，而不是一开始猜的"下游服务性能差"。

3.2 第一轮误判：所有人都以为是模型服务扛不住

晚上八点开始压测，刚开始 10 分钟一切正常，响应时间稳定在 250 毫秒左右，错误率 0%。我当时还在跟同事说，这轮应该稳了。到了第 18 分钟，聚合报告里的 99% 线开始往上翘，从 300 毫秒跳到 900 毫秒。第 25 分钟，错误率爬到 5%，报错内容是"AI 推理服务调用超时"。

几乎所有人的第一反应都是——AI 模型服务的并发上限不够了，要扩容。当时值班的同事还真的去 AI 服务那边加了两个实例，加了之后错误率短暂降下来一点，但没持续多久又反弹回去。这个现象其实很典型：下层加了资源，但上层进来的流量是不受控的，加多少都能把你的带宽填满。如果你是靠扩容来"解决问题"，本质上是在给失控的上游填坑，永远填不满。

我这个时候已经在怀疑限流层没起作用，但手里没有证据。单纯的日志里只看到"调用超时"，没有任何信号量相关的统计信息。我当时最大的失误是，给信号量这块没加监控埋点，导致排查时没有第一手数据支撑，全靠推理。

3.3 凌晨两点，线程栈和 hashCode 日志指向了同一结论

转机出现在我决定给代码临时加日志：在 Semaphore 的 acquire 之前打印semaphore.hashCode()和当前活跃线程数，重新打包部署到压测环境，继续压。日志刚刷出来我就意识到情况不对——20 个并发请求，打出来的 hashCode 至少有 10 个不同的值，而且同一个请求在不同重试里的 hashCode 也不一样，说明信号量在请求生命周期里被多次重建了。后面再看线程栈，发现完全没有任何线程在Semaphore.acquire()上等待，所有请求一路畅通无阻地涌向下游 AI 服务。

那一刻算是彻底破案了。Semaphore 没在类加载时初始化，而是在Flux.defer的 Supplier 里每次订阅都新创建，所有请求各拿各的信号量，限流只是背了个名字，并没有真的下单。线程栈里看不到等待，是因为压根没有公共闸门。一个写错位置的 Semaphore，让整条"并发控制"变成了一场自我安慰。

凌晨两点二十三分，我把定位结果发到团队群，顺手改了一版代码，核心改动只有一行——把 Semaphore 提到常量定义区，变成全局单例。重新跑压测，50 并发，只让 10 个同时进 AI 服务，响应时间立刻回到 230 毫秒，错误率归零。一场闹剧，最后被一行代码收场。

4. 生产级修复：Semaphore 到底应该放在哪一层

4.1 三种修复思路逐一对比

修复方式其实不只有一种，我对比了三个方案再选的最终版：

方案 A：Semaphore 提到成员变量/静态变量位置。这是最直接的修复，把信号量定义在类字段上，defer 里只引用同一个实例。改造成本极小，相当于把 new 的动作从每次订阅挪到类加载时执行一次。这个方案保留了原来的设计意图，即每个请求走独立的 Flux 数据流，但共享同一个闸门。

public class AgentInvoker { private static final Semaphore AI_SEMAPHORE = new Semaphore(10); public Flux<String> invokeAgent(AgentRequest request) { return Flux.defer(() -> Mono.fromCallable(() -> agentService.predict(request)) .doFirst(() -> AI_SEMAPHORE.acquire()) .doFinally(signal -> AI_SEMAPHORE.release()) ); } }

方案 B：不用 Flux.defer，直接用 Mono.fromCallable + flatMap 的 maxConcurrency 控制。这个方案从"并发数"层面限制流量，能在响应式链路的 flatMap 阶段直接限制同时执行的任务数，不需要显式信号量。但它的局限在于，maxConcurrency 只能控制当前 flatMap 里任务的并行度，如果你的并发源分布在不同入口方法里，各自都有 flatMap，那总并发还是无界的。适合单一入口的简单场景，不适合我们这种多触发源的 WMS 网关。

方案 C：用 Resilience4j 的 Bulkhead 或 RateLimiter 包装调用。这是把限流抽成框架能力，代码侵入少，但多了一层依赖和一个额外配置项。对于 AI 集成这个局部场景来说，引入整套容错框架有一点"杀鸡用牛刀"的感觉，而且跟响应式链路的适配还需要额外注意。后续如果系统里容错场景变多，这个方向是值得迁移的，但现在保持简单更好。

最终选了方案 A。理由很实际：改动最小、贴合现有代码结构、不需要引入新依赖，而且能在最贴近调用点的位置控制流量入口。

4.2 修复后的完整代码示例与细节解释

生产上最终落地的完整写法如下：

@Component public class AgentInvocationGateway { private static final int AI_MAX_CONCURRENCY = 10; private static final Semaphore AI_SEMAPHORE = new Semaphore(AI_MAX_CONCURRENCY); private final AgentService agentService; public AgentInvocationGateway(AgentService agentService) { this.agentService = agentService; } public Flux<String> invokeAgent(AgentRequest request) { return Flux.defer(() -> { log.debug("Agent invoke start, semaphore={}, availablePermits={}", AI_SEMAPHORE.hashCode(), AI_SEMAPHORE.availablePermits()); return Mono.fromCallable(() -> agentService.predict(request)) .doFirst(() -> acquireSafely()) .doFinally(signal -> AI_SEMAPHORE.release()); }).onErrorResume(error -> { log.error("Agent invoke failed: request={}, error={}", request, error.getMessage()); return Flux.empty(); }); } private void acquireSafely() { try { if (!AI_SEMAPHORE.tryAcquire(3, TimeUnit.SECONDS)) { log.warn("Agent concurrency limit exceeded, wait timeout"); throw new AgentOverloadException("AI 服务忙，请稍后重试"); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new AgentOverloadException("并发控制中断"); } } }

这段代码有几个细节值得说一下。

第一，tryAcquire(3, TimeUnit.SECONDS)比acquire()更友好。如果 AI 服务持续过载，请求会快速失败并抛出明确异常，而不是无限期阻塞在信号量上把线程池耗尽。对仓储场景来说，一个拣货指令快速失败可比让它排队等 30 秒重要得多——仓库作业人员可以直接重扫，但干等会让人崩溃。

第二，onErrorResume做了兜底。AI 调用失败不能影响主业务流程，WMS 里任何异常如果直接抛给上游，RF 枪可能直接白屏，这是绝对不能接受的。失败后降级为返回空流，业务层可以继续走人工判断逻辑。

第三，日志里打印availablePermits()是个好习惯。线上排查任何限流问题的时候，有一条信号量余量日志会让你快速判断是不是这里出问题了，而不至于又抓一晚上线程栈。

4.3 生产部署时还要注意的另外三个点

修复完信号量之后，我们趁压测环境还在，顺手把生产部署的另外几个并发安全相关问题也做了加固，这三个点也很值得提一下。

第一个是flatMap的maxConcurrency参数不能乱设。我们之前有个内部服务编排的flatMap没配这个参数，默认是 256。这意味着哪怕信号量控制了 AI 服务的入口并发，编排层内部的并行度也可能过高，导致线程切换开销巨大。所有响应式链路上的flatMap，特别是涉及外部调用的，都建议显式指定并发度，常规 8 到 32 比较合适。

第二个是 ThreadLocal 的传递问题。WebFlux 的线程模型是非绑定的，一个请求可能在不同的 reactor 线程上执行，直接往 ThreadLocal 里塞登录用户或者追踪 ID 会在异步边界丢失。我们用 Reactor Context 来传递 traceId，配合日志组件的 MDC 适配器，这样每次请求的日志能串成一条链路，排查问题的时候能省一半时间。

第三个是监控告警要针对"等待中线程数"而不是"错误率"设置阈值。错误率高说明已经出大事了，等待中线程数开始增长说明限流正在生效，这是事故的前兆，应该在信号量触发告警前就发现流量异常。我们在 Micrometer 里注册了一个 Gauge，用来暴露AI_SEMAPHORE.getQueueLength()，超过 20 就发钉钉告警。这个指标在生产上是真正的"金丝雀"。

5. 常见问题排查速查表与几点总结体会

5.1 排查速查表：以后再遇到类似问题，直接对着查

这张表基本上覆盖了我们从问题出现到修复过程中的全部排查要点，对正在做 WebFlux 响应式架构并且接第三方服务的团队都有参考意义。

5.2 这次踩坑给我们的三点运维理念冲击

第一个理念冲击是"资源对象绝不能放在按订阅执行的工厂代码里创建"。不止 Semaphore，连接池、限流器、计数器、锁对象，凡是需要跨请求共享的，都必须跟订阅生命周期解耦。这是一条比"不要用全局变量"更现代的警戒线，因为 Reactor 的订阅模型让这个问题变得非常隐蔽。

第二个冲击是"压测务必跑破坏性场景"。如果只在常规并发下验证功能，很多并发安全问题根本不会显现。信号量这个坑在 20 并发以下怎么压都是绿的，一上 50 并发持续运行，问题立刻现形。Jmeter 压测时记得持续运行至少 30 分钟，观察 99% 响应时间的漂移趋势，这是发现并发安全隐患的最有效手段。

第三个冲击是"监控指标要贴着资源设计"。如果你的核心资源是 Semaphore，那它就是最高优先级的监控对象，队列长度、等待线程数、许可余量都得暴露出来。否则哪怕你有全链路追踪，也只能像我们一样从下游超时倒推上游限流失效，效率低得吓人。

Swift 那边最近也在聊并发安全，Swift 的 Actor 模型通过隔离可变状态从源头消除数据竞争，跟 Java 体系里用 Semaphore 控制临界区的思路是两个方向。但底层逻辑是相通的——并发安全的核心永远是:问清楚"谁拥有这个资源、谁负责它的创建和销毁、哪些路径可以碰它"。你把资源的所有权写清楚，问题就解决了一半。

5.3 最后一句话分享

我始终觉得，生产环境里 90% 的诡异问题，最终都不是什么高深莫测的玄学，而是一个看似合理的代码放错了位置。Semaphore 放在 Flux.defer 里这件事，写代码的时候毫无违和感，压测之前没有任何人会盯着它看。但恰恰是这种"看起来没毛病"的代码，最容易在夜深人静的时候给你上一课。

这一课我们上得很值。现在这套 WMS 的 AI 集成链路已经稳定跑了两个多月，信号量一直是那个全局单例，tryAcquire的超时保护也在发挥作用，压测报告里 99% 响应时间的曲线平滑得像心跳监测器。如果你也在做类似的响应式集成，听我一句劝：先去检查你的 Semaphore 是不是唯一实例，再上压测，别等到凌晨三点才后悔。