如何在Spring WebFlux（Netty服务器）中限制并发请求数

我想实现类似Tomcat中设置server.tomcat.max-threads=100的效果——通过设置线程池大小来限制最大并发处理请求数为100，但在使用Netty作为服务器的Spring WebFlux中，找不到对应的配置属性。

我已尝试以下方案：

• 定义ReactiveWebServerFactory自定义Bean，设置NioEventLoopGroup线程数为2，但测试时发现同一时间只能处理一个请求（即使是不同接口），其他请求需等待前一个完成：

@Bean
public ReactiveWebServerFactory reactiveWebServerFactory() {
        NettyReactiveWebServerFactory factory = new NettyReactiveWebServerFactory();
        factory.addServerCustomizers(builder -> builder.runOn(new NioEventLoopGroup(2)));

        return factory;
}

• 设置server.netty.max-keep-alive-requests=2属性，但仍能处理超过2个并发请求；
• 设置server.tomcat.max-threads=2，如预期无效果。

测试用API代码：

@RestController
public class CheckController {

    @PostMapping("/test")
    public Mono<Long> something() throws InterruptedException {
        System.out.println("cores " + Runtime.getRuntime().availableProcessors());
        return Mono.just(2L)
                .flatMap(d -> simulateBlockingOperation())
                .log();
    }

    @PostMapping("/test2")
    public Mono<Long> something2() throws InterruptedException {
        return Mono.just(1L)
                .log();
    }

    private Mono<Long> simulateBlockingOperation() {
        System.out.println("Current thread " + Thread.currentThread().getName());
        int x = 0;
        while(x!=1) {}
        return Mono.just(2L);
    }
}

临时解决方法：创建自定义线程池并在每个接口用.subscribeOn切换线程，但这种方法需要修改每个接口，不够优雅：

@RestController
public class CheckController {

    private final Scheduler customThreadPool;

    public CheckController(Scheduler customThreadPool) {
        this.customThreadPool = customThreadPool;
    }
    
    @Bean
    public Scheduler reactiveRequestThreadPool() {
       return Schedulers.newBoundedElastic(2, 2, "my-custom-thread");
    }

    @PostMapping("/test")
    public Mono<Long> something() throws InterruptedException {
        return Mono.just(2L)
                .doOnNext(d -> getCurrentThread())
                .flatMap(d -> simulateBlockingOperation())
                .subscribeOn(customThreadPool)
                .log();
    }

    @PostMapping("/test2")
    public Mono<Long> something2() throws InterruptedException {
        return Mono.just(1L)
                .doOnNext(d -> getCurrentThread())
                .subscribeOn(customThreadPool)
                .log();
    }

    private Mono<Long> simulateBlockingOperation() {
        int x = 0;
        while(x!=1) {}
        return Mono.just(2L);
    }

    private void getCurrentThread() {
        System.out.println("current thread " + Thread.currentThread().getName());
    }
}

请问在Spring WebFlux中是否可以优雅地限制最大并发请求数？

核心原理说明 #

你之前遇到的问题，本质是混淆了Netty IO线程（EventLoop） 和 业务处理线程 的职责：

• Netty的EventLoop线程是用来处理IO操作（如请求接收、响应发送）的，不能执行阻塞任务，否则会阻塞整个IO线程，导致其他请求无法处理；
• 要限制业务请求的并发数，需要将业务逻辑（尤其是阻塞操作）放到专门的业务线程池中执行。

优雅的全局并发限制方案 #

推荐通过全局WebFilter统一将所有请求的业务处理切换到自定义的有界弹性线程池，无需修改每个Controller接口：

@Configuration
public class WebFluxConcurrencyConfig {

    // 配置有界弹性线程池：核心线程数100，任务队列容量1000，线程前缀webflux-concurrency
    @Bean
    public Scheduler boundedElasticScheduler() {
        return Schedulers.newBoundedElastic(100, 1000, "webflux-concurrency");
    }

    // 全局WebFilter，将所有请求的业务处理切换到自定义线程池
    @Bean
    public WebFilter concurrencyLimitFilter(Scheduler boundedElasticScheduler) {
        return (exchange, chain) -> chain.filter(exchange)
                .subscribeOn(boundedElasticScheduler);
    }
}

参数说明：

• newBoundedElastic(100, 1000, ...)：
  • 第一个参数：核心线程数，即同时处理的最大并发请求数；
  • 第二个参数：任务队列的最大容量，当并发请求数超过核心线程数时，新请求会进入队列等待；
  • 当队列也满了之后，新的请求会被直接拒绝（抛出RejectedExecutionException），达到严格限制并发的效果。

补充：Netty连接数限制（非请求并发） #

如果你需要限制的是TCP连接数而非请求并发数，可以通过NettyReactiveWebServerFactory配置TCP参数，但这和Tomcat的max-threads不是同一个概念：

@Bean
public ReactiveWebServerFactory reactiveWebServerFactory() {
    NettyReactiveWebServerFactory factory = new NettyReactiveWebServerFactory();
    factory.addServerCustomizers(server -> 
        server.tcpConfiguration(tcp -> 
            tcp.option(ChannelOption.SO_BACKLOG, 100) // 监听队列的最大长度
        )
    );
    return factory;
}

为什么之前的方案无效？ #

1. 自定义NioEventLoopGroup：你设置的是IO线程数，但你的simulateBlockingOperation是无限阻塞操作，直接占住了IO线程，导致其他请求无法被处理；
2. server.netty.max-keep-alive-requests：这个配置是单个长连接上能处理的最大请求数，和并发请求数无关；
3. server.tomcat.max-threads：WebFlux用Netty时，Tomcat的配置自然无效。

内容的提问来源于stack exchange，提问作者Patrick