如何用Spring WebFlux创建带背压控制的循环以避免内存溢出？

针对你需要从repository分批拉取数据、生成支持背压的Flux的需求，这里提供两种简洁可靠的实现方式：

方式一：使用Flux.expand（推荐） #

Flux.expand 可以递归生成流，每处理完当前批次后再请求下一批，天然适配背压逻辑，代码更简洁：

阻塞式Repository场景（如JDBC） #

如果你的repository.fetch()是阻塞同步方法，需要将其包装到异步线程池中，避免阻塞Reactor的IO线程：

// 初始化第一批数据的异步调用
Mono<List<Object>> firstBatch = Mono.fromCallable(repository::fetch)
    .subscribeOn(Schedulers.boundedElastic());

Flux<List<Object>> batchFlux = firstBatch.flux()
    // 递归获取下一批，直到返回空列表
    .expand(batch -> {
        if (batch.isEmpty()) {
            return Mono.empty();
        }
        return Mono.fromCallable(repository::fetch)
            .subscribeOn(Schedulers.boundedElastic());
    })
    // 过滤掉最后一次返回的空列表
    .filter(batch -> !batch.isEmpty());

响应式Repository场景（如R2DBC） #

如果你的repository本身是响应式的（fetch()返回Mono<List<Object>>），可以直接简化：

Flux<List<Object>> batchFlux = repository.fetch()
    .expand(batch -> batch.isEmpty() ? Mono.empty() : repository.fetch())
    .filter(batch -> !batch.isEmpty());

方式二：使用Flux.generate（状态化控制） #

Flux.generate 支持通过状态跟踪控制生成逻辑，每次仅在下游请求元素时才执行拉取操作，天然支持背压：

Flux<List<Object>> batchFlux = Flux.generate(
    // 初始化状态：标记是否还有数据可拉取
    () -> true,
    (hasMoreData, sink) -> {
        if (!hasMoreData) {
            sink.complete();
            return false;
        }

        List<Object> batch = repository.fetch();
        if (batch.isEmpty()) {
            sink.complete();
            return false;
        } else {
            sink.next(batch);
            // 继续保留可拉取状态
            return true;
        }
    }
);

注意：如果fetch()是阻塞方法，同样需要将其包装到Mono.fromCallable().subscribeOn(...)中，避免阻塞Reactor线程。

你之前方案的问题分析 #

1. Flux.generate内存耗尽：
   大概率是因为repository.fetch()是阻塞操作且未放到异步线程池，导致Reactor的事件线程被阻塞，无法正确响应下游的背压信号，进而持续拉取数据填满内存。

2. Flux.create实现复杂：
   这种场景下完全没必要用Flux.create，它更适合处理外部异步事件（如回调、消息队列），而批量拉取数据用expand或generate更贴合需求，代码也更简洁可控。

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