Spring WebFlux 怎样限制并发请求的背压范围

Spring WebFlux 要限制并发请求，必须先明确范围：flatMap(mapper, concurrency, prefetch)约束一条订阅链中同时活跃的内部序列；多个 HTTP 请求共用的下游容量，需要另设共享准入。背压传递的是元素需求，不能自动变成全站并发上限、每秒请求配额或总内存预算。

突增请求为什么会穿透单条链的并发设置

假设一个聚合接口接收一批商品编号，再通过 WebClient 查询详情。排查时看到的时间线可能是：单请求访问正常，入口并发增加后，下游等待时间变长，随后出现超时。这里是用于推理的场景，并非一份已执行的生产事故报告。

根因线索在于“每个请求各自订阅”。一条链设置并发 6，十条同时活跃的链在资源允许时就可能合计订阅 60 个内部调用。这个算式描述范围叠加，不是实测吞吐，也没有计入连接限制和请求完成速度。把参数 6 当成整个服务最多六个下游请求，修复就会落错位置。

排查第一步是区分入口活跃请求数、每条链的内部订阅数、服务实例的下游活跃调用数。三者分别记数，才能判断压力增加发生在哪一层。

先给每次订阅设置明确的并发上限

// 每次订阅最多维持 6 个活跃的内部序列
Flux values = ids.flatMap(this::fetch, 6, 1);

这里的 ids 是编号流，fetch 返回一个表示单次远程查询的 Mono。只有真正受订阅驱动的调用才能被这个组合约束；如果在组装链之前就启动任务，或者在 mapper 内手动调用 subscribe()，任务生命周期便脱离了这条链的控制。

concurrency控制同时活跃的内部序列数量；prefetch控制对每个内部序列预请求的元素规模。对于只返回一个值的 Mono，prefetch 不是主要容量旋钮。把它写为 1，既不表示总共只缓存一个元素，也不表示只允许一个 HTTP 请求。

limitRate拆分向上游传播的需求批次。它可以帮助调整某个位置的预取行为，但没有时间窗口，不能表达“每秒六次”，也不能替代跨订阅的共享并发准入。需求沿操作符链传播，途中还可能被其他操作符重新塑形。

本步应核对 mapper 是否只返回 Publisher、远程调用是否延迟到订阅时启动，以及是否有显式缓存、提前收集为列表或额外的手动订阅。flatMap 会交错返回结果，若业务需要保持编号顺序，还必须单独选择排序策略并考虑对应缓冲代价。

把共享下游容量放到单例准入层

如果约束目标是“这个应用实例同时最多持有 24 个下游调用许可”，可以把许可放进 Spring 单例服务。以下示例采用立即拒绝策略，没有增加等待队列。数字只是示例，需要按照下游承载能力与实例数量配置。

import java.time.Duration;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.function.Supplier;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Service
public class DetailService {
    private final WebClient client;
    // 此对象随单例服务共享，不能为每个请求重新创建
    private final Semaphore permits = new Semaphore(24);

    public DetailService(WebClient client) {
        this.client = client;
    }

    public Flux fetchAll(Flux ids) {
        // 同时保留每个订阅的局部限制
        return ids.flatMap(this::fetch, 6, 1);
    }

    private Mono fetch(String id) {
        return guarded(() -> client.get()
                .uri("/items/{id}", id)
                .retrieve()
                .bodyToMono(String.class)
                .timeout(Duration.ofSeconds(2)));
    }

    private  Mono guarded(Supplier> operation) {
        return Mono.defer(() -> {
            // 不在事件循环上阻塞等待许可
            if (!permits.tryAcquire()) {
                return Mono.error(new RejectedExecutionException("下游容量已满"));
            }
            // 供应器抛错也转为错误信号，并经过统一释放路径
            return Mono.defer(operation)
                    .doFinally(signal -> permits.release());
        });
    }
}

项目需要提供已配置 baseUrl 的 WebClient Bean，并使用与当前 Spring 项目兼容的 Reactor 版本。示例没有创建 Controller；在现有响应式入口中返回 fetchAll的结果即可，不要为了取结果调用 block()。

外层 Mono.defer保证每次订阅重新尝试取得许可。内层 defer 将供应器调用纳入信号生命周期；doFinally处理完成、错误与取消后的许可释放。不要使用阻塞的 acquire()，否则事件循环可能被等待许可占住。

这一实现的边界是单个服务对象，通常对应单个应用实例。多实例部署时，各实例各有 24 个许可；如果需要所有实例共享的下游总配额，还要在部署配额、网关或共享协调层分配容量。

立即拒绝会改变错误行为。普通 flatMap 中一个内部错误通常会使整体链失败，并取消其他内部订阅。应在接口错误映射层将容量不足转换成明确的服务繁忙响应，例如按接口契约返回 503；不要把内部异常文本直接展示给调用方。如果接口允许逐项成功或失败，应另行设计逐项结果对象，不能用空值偷偷丢掉失败项。不要对容量拒绝立即无限重试。

复查队列、响应体与取消后的资源占用

并发下降不等于所有积压都消失。连接池等待队列、上游已经收集的编号列表、编码器缓冲以及下游实际执行的工作，都有自己的生命周期。对字符串响应，bodyToMono(String.class)会聚合响应内容，需同时配置合理的编解码缓冲上限：

// 构建项目所用 WebClient 时设置，baseUrl 仍由项目配置
WebClient client = WebClient.builder()
        .baseUrl(remoteBaseUrl)
        .codecs(c -> c.defaultCodecs().maxInMemorySize(64 * 1024))
        .build();

这里的 64 KiB 是示例编解码缓冲配置，不是整个进程的内存上限，也不适用于无条件接收任意大文档。超过限制可能产生 DataBufferLimitException，应按实际响应协议处理。若响应需要真正流式消费，应采用适合协议的流式解码方式，并避免末尾再次聚合全部内容。

超时后远程任务一定停止吗？不一定。本地 timeout 会终止当前链路等待并触发取消，许可会释放，但远程服务可能继续处理已经收到的请求。这里限制的是本地逻辑调用许可，不能据此证明远程物理任务也立即减少。需要结合服务端取消支持、请求幂等性和下游超时策略控制残留工作。

修复后应观察什么？在隔离环境分别增加单次请求的编号数量和同时请求数量，观察每订阅活跃内部序列、共享许可占用、准入拒绝数、连接等待数、超时数和响应体大小分布。预期局部并发不超过设定值，共享许可不会超过配置容量；拒绝或超时可能增加，不能把错误被快速返回解释为吞吐已经提升。本文未执行压测，以上是读者可采用的复查动作。

还应加入取消与供应器抛错场景，确认许可最终归还；避免在服务外继续持有已失效的订阅。局部并发、共享准入和队列预算共同明确后，背压的作用范围才有可解释的边界。

官方参数说明可复制查阅：https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html；Mono 生命周期：https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Mono.html；WebClient 缓冲配置：https://docs.spring.io/spring-framework/reference/web/webflux-webclient/client-builder.html；WebFlux 说明：https://docs.spring.io/spring-framework/reference/web/webflux.html。