使用ParallelFlux并行执行任务时如何限制线程数并解决阻塞问题？

嗨，我来帮你梳理下当前遇到的问题：你的定时任务执行时会阻塞整个应用，导致其他请求（比如Postman的接口调用）无法响应，而且设置parallel(3)也没达到预期效果。咱们先找问题根源，再给出具体解决方案。

问题根源分析 #

1. 调度线程被阻塞：@Scheduled和@PostConstruct注解的方法默认是在Spring的调度器单线程上执行的。如果你的fillMap方法一直等待Flux执行完成（比如隐式同步等待），就会把这个调度线程占死，导致其他操作无法获得线程资源。
2. 并行策略与线程池不匹配：你用了Schedulers.parallel()，这个线程池是为CPU密集型任务设计的，而你的getData方法是HTTP请求（IO密集型），用这个线程池容易耗尽线程，且没有正确异步化整个流程。
3. WebClient重复创建：在getData里每次新建WebClient实例，会造成不必要的资源开销，影响性能。

具体解决方案 #

方案一：用parallel() + runOn()指定并行数并异步执行 #

修改fillMap方法，重点是异步订阅（不阻塞调度线程），并选择适合IO密集型的线程池：

@Scheduled(fixedRate = 60000L) 
@PostConstruct 
public void fillMap() {
    Flux.fromIterable(proxyParserService.getProxyList())
        .parallel(3) // 明确指定并行数为3
        .runOn(Schedulers.boundedElastic()) // IO密集型任务用boundedElastic，避免线程耗尽
        .flatMap(geoDataService::getData)
        // 这里放你的业务逻辑...
        .subscribe(); // 异步订阅，方法立即返回，不阻塞调度线程
}

方案二：用flatMap的并发参数控制（更简洁的IO密集型方案） #

如果是IO密集型任务，其实不需要parallel()，直接用flatMap的第二个参数指定最大并发数即可：

@Scheduled(fixedRate = 60000L) 
@PostConstruct 
public void fillMap() {
    Flux.fromIterable(proxyParserService.getProxyList())
        .flatMap(geoDataService::getData, 3) // 第二个参数设置最大并发数为3
        // 这里放你的业务逻辑...
        .subscribeOn(Schedulers.boundedElastic())
        .subscribe(); // 异步订阅，释放调度线程
}

额外优化：WebClient复用 #

不要在getData里每次新建WebClient，建议注入WebClient.Builder来复用实例：

// 在GeoDataService中注入WebClient.Builder
private final WebClient.Builder webClientBuilder;

public GeoDataService(WebClient.Builder webClientBuilder) {
    this.webClientBuilder = webClientBuilder;
}

public Mono<Address> getData(Address proxy) {
    return webClientBuilder.build()
        .get()
        .uri(String.format(URL, proxy.getHost())) // 直接动态指定URI，不用每次新建WebClient
        .retrieve()
        .onStatus(HttpStatus::isError, clientResponse -> {
            log.error("Error while calling endpoint {} with status code {}", URL, clientResponse.statusCode());
            return Mono.error(new RuntimeException("Error while calling geolocation endpoint")); // 用Mono.error抛出异步错误
        })
        .bodyToMono(Address.class);
}

关键注意事项 #

• 绝对不要在@Scheduled方法中用block()：block()会强制同步等待结果，直接阻塞调度线程，这是导致你应用无法响应其他请求的核心原因。
• 线程池选择要匹配任务类型：CPU密集型用Schedulers.parallel()，IO密集型用Schedulers.boundedElastic()，避免线程资源耗尽。
• 异步订阅是核心：用subscribe()代替同步等待，让调度线程立即释放，处理其他请求。

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