如何用WebFlux并发调用分页API并异步处理图片下载？

1. 为什么forEach语句从未执行？ #

WebFlux的响应式流是懒加载的，只有当调用subscribe()方法订阅流之后，整个数据流才会开始执行。你当前的downloadImages()方法只构建了Flux，但没有触发订阅，所以整个流水线完全没启动，自然response.getImages().forEach(...)不会执行。

2. 什么是Publisher？ #

Publisher是Reactive Streams规范的核心接口，Flux和Mono都是它的实现类。它代表一个可以异步发射数据的数据源，只有当有订阅者（Subscriber）订阅它时，才会开始发射数据。简单来说，它就是响应式编程里"数据流"的抽象，所有响应式操作（比如flatMap、map）都是基于它构建的。

3. downloadImages方法的正确实现 #

首先要修正几个核心问题：

• 方法名笔误：fetchPages里调用的是downloadImage，但你定义的是downloadImages，需要统一命名
• 不能返回null：返回null会导致响应式流抛出空指针异常，必须返回合法的Publisher（比如Mono.empty()表示操作完成）
• 阻塞IO操作要切换线程：下载图片属于阻塞IO，不能在WebFlux的Netty事件循环线程执行，要切换到专门的阻塞IO线程池（Schedulers.boundedElastic()）

以下是修正后的完整代码：

@Service
@Slf4j
public class MyService {

    private final WebClient webClient;
    // 专门处理阻塞IO的线程调度器
    private final Scheduler ioScheduler = Schedulers.boundedElastic();

    // 用构造注入WebClient，符合Spring最佳实践
    public MyService(WebClient.Builder webClientBuilder) {
        this.webClient = webClientBuilder.baseUrl("http://host").build();
    }

    public void downloadImages() {
        Flux.range(1, 5) // 直接生成1-5的页码流，无需转List
                .flatMap(this::fetchPages, 10) // 并发10个请求获取分页数据
                .subscribe(); // 必须调用subscribe触发整个流执行
    }

    private Flux<ImageResponse> fetchPages(int page) {
        // 假设API返回单个APIResponse对象（包含images列表），用bodyToMono
        return webClient.get()
                .uri("/images?page={page}", page)
                .retrieve()
                .bodyToMono(APIResponse.class)
                .flatMapMany(apiResponse -> Flux.fromIterable(apiResponse.getImages())) // 把图片列表拆成单个ImageResponse的流
                .flatMap(this::downloadImage); // 处理每张图片的下载
    }

    // 修正方法名，返回Mono<Void>表示无返回值的异步操作
    private Mono<Void> downloadImage(ImageResponse image) {
        // 将阻塞的下载逻辑包装到ioScheduler线程池
        return Mono.fromRunnable(() -> {
            try {
                // 这里替换为实际的图片下载逻辑，比如通过URL下载并保存到本地
                log.info("开始下载图片: {}", image.getImageURL());
                // 示例：模拟下载耗时
                Thread.sleep(100);
                log.info("完成下载图片: {}", image.getImageURL());
            } catch (Exception e) {
                log.error("下载图片失败: {}", image.getImageURL(), e);
            }
        }).subscribeOn(ioScheduler) // 指定阻塞操作在io线程池执行
          .then(); // 返回Mono<Void>标记操作完成
    }
}

补充说明：如果你的API返回的是APIResponse的列表（而非单个包含图片列表的对象），请将bodyToMono(APIResponse.class)改为bodyToFlux(APIResponse.class)，其余逻辑不变。

4. parallelStream vs 线程池：选哪个？ #

绝对不要用parallelStream，原因如下：

• parallelStream依赖JVM共享的ForkJoinPool.commonPool()，阻塞操作过多会耗尽该线程池，影响其他任务执行
• WebFlux的响应式模型依赖非阻塞事件循环线程，parallelStream的阻塞操作会破坏非阻塞特性，导致系统性能下降甚至崩溃

正确的做法是使用WebFlux提供的Schedulers.boundedElastic()，它会自动管理线程池，专门处理阻塞IO操作，不会阻塞Netty事件循环，保证响应式系统的稳定性和性能。

内容的提问来源于stack exchange，提问作者alejo.montania