响应式编程中Semaphore的坑：别把它放进Flux.defer

前阵子我们在给一套 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 FluxString 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 做持续压测配置其实不复杂但有几个细节直接影响结论质量。按下述方式配基本能压出生产问题的真实面貌。第一步新建线程组设置线程数为 50Ramp-up 时间为 5 秒循环次数设置为无限并勾选调度器持续运行 30 分钟。Ramp-up 太短会导致瞬间流量冲击把问题放大为系统崩溃而掩盖了限流失效的真相Ramp-up 太长又达不到并发高峰的测试目标。5 秒内把 50 个线程全部拉起是比较合理的渐进式加压方式。第二步添加 HTTP 请求取样器配置我们内部 AI Agent 网关接口的路径和参数。注意这里参数最好用一个 CSV 数据集配置每个线程用自己的请求数据避免所有线程拿着同一份参数压同一个缓存 key。当时我们第一批压测脚本偷懒用了固定参数结果命中缓存之后响应时间全是 1 毫秒什么问题都测不出来。第三步添加聚合报告和查看结果树两个监听器另外强烈建议加一个jpgc - 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 三种修复思路逐一对比修复方式其实不只有一种我对比了三个方案再选的最终版方案 ASemaphore 提到成员变量/静态变量位置。这是最直接的修复把信号量定义在类字段上defer 里只引用同一个实例。改造成本极小相当于把 new 的动作从每次订阅挪到类加载时执行一次。这个方案保留了原来的设计意图即每个请求走独立的 Flux 数据流但共享同一个闸门。public class AgentInvoker { private static final Semaphore AI_SEMAPHORE new Semaphore(10); public FluxString 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 FluxString 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 排查速查表以后再遇到类似问题直接对着查现象可能原因排查动作解决方案并发高时信号量形同虚设接口却全部成功Semaphore 被放在 Flux.defer 工厂里重复创建打印信号量 hashCode观察是否随请求变化将 Semaphore 提升为静态/实例字段保证全局单例请求在 AI 服务处大量超时错误率攀升限流失效导致模型服务过载观察线程栈是否有线程阻塞在 acquire 上修复信号量同时设置 flatMap 并发度调用了 tryAcquire 但大量请求仍然成功acquire 失败被吞掉异常没有向上抛出确认 onErrorResume 是否吞掉 AgentOverloadException区分业务兜底与系统错误错误要打日志响应时间飙升但 CPU 利用率不高线程大量阻塞在信号量等待event loop 周期变长查看线程 dump关注 WAITING 状态线程使用 tryAcquire 超时快速失败降低等待时长日志串不起一次请求的完整链路WebFlux 非绑定线程导致 ThreadLocal 丢失检查日志中 traceId 是否为空用 Reactor Context 传递 traceId配合 MDC 适配这张表基本上覆盖了我们从问题出现到修复过程中的全部排查要点对正在做 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 是不是唯一实例再上压测别等到凌晨三点才后悔。