升级spring-boot-starter-webflux至2.4.5后reactor-core线程挂起问题咨询

故障本质说明 #

你观测到AbstractQueuedSynchronizer.tryAcquireShared返回-1后线程挂起，本质是代码末尾调用collectList().block()时，依赖AQS共享锁实现等待逻辑，返回-1代表对应Flux流一直没有完成（既没有正常输出所有元素，也没有抛出异常终止），导致阻塞线程一直处于等待状态。

版本差异导致故障的几个核心原因 #

• Spring Boot 2.4.x对应Spring Framework 5.3.x、Reactor Netty 1.0.x，和2.3.x版本的核心组件行为有两个关键变化：
  1. 原Schedulers.elastic()无界弹性线程池被废弃，默认指向新增的Schedulers.boundedElastic()，该线程池有最大线程数、任务队列容量上限，默认参数对高并发场景不友好，任务堆积时会出现等待，不会自动创建新线程处理
  2. Reactor Netty默认连接池的pendingAcquireTimeout默认值调整为无限等待，当WebClient并发请求数超过默认连接池最大连接数（默认值为CPU核心数*2）时，新增请求会一直等待获取连接，不会抛出超时异常，导致流一直处于pending状态
• 你的代码中只捕获了WebClientResponseException，连接超时、连接池获取等待超时、网络异常等其他异常没有兜底处理，部分异常场景下可能出现流未正常终止的问题
• 多余的Flux.merge(flx)操作虽然不是核心故障原因，但会带来不必要的订阅开销

修复方案 #

1. 显式配置WebClient连接池，避免无限等待 #

自定义WebClient底层的HttpClient连接池，明确配置连接数、获取连接超时、响应超时参数：

HttpClient httpClient = HttpClient.create(ConnectionProvider.builder("webclient-pool")
        // 按业务实际并发量调整最大连接数
        .maxConnections(200)
        // 配置获取连接的超时时间，超过10秒直接抛异常，避免无限等待
        .pendingAcquireTimeout(Duration.ofSeconds(10))
        .build())
        // 配置接口响应超时
        .responseTimeout(Duration.ofSeconds(10));

WebClient webClient = WebClient.builder()
        .clientConnector(new ReactorClientHttpConnector(httpClient))
        .build();

2. 替换默认线程池，自定义弹性调度器 #

不要使用默认的Schedulers.elastic()/Schedulers.boundedElastic()，按业务需求自定义调度器：

// 参数依次为最大线程数、队列最大容量、线程名前缀
Scheduler customScheduler = Schedulers.newBoundedElastic(300, 10000, "api-call-scheduler");

// 业务代码中使用自定义调度器
Flux<String> flx = Flux.fromIterable(DetailsList)
        .parallel()
        .runOn(customScheduler)
        .flatMap(this::getDetail)
        .sequential();

3. 补全异常兜底逻辑 #

在WebClient调用的onErrorResume中捕获所有异常，避免异常漏处理导致流挂起：

.onErrorResume(ex -> {
    log.error("调用接口失败，序列号：{}", requestDetail.getSerial(), ex);
    return Mono.just(requestDetail.getSerial() + "_INTERFACE_ERROR");
});

4. 优化冗余代码 #

去掉多余的Flux.merge调用，直接对已经生成的Flux流做收集操作：

// 优化前
// List<String> allRawResults = Flux.merge(flx).collectList().block();
// 优化后
List<String> allRawResults = flx.collectList().block();

5. 长期优化建议 #

尽量避免在响应式代码中使用block()阻塞操作，将方法返回值改为Mono<Response>，全链路非阻塞运行，从根源上避免阻塞死锁问题。

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