如何在Reactor中实现首次调用返回有效值后终止后续WebFlux请求

实现方案 #

你要的顺序执行IO、命中第一个就终止的逻辑可以通过concatMap+filter+next的组合实现，完全匹配原有同步代码的执行逻辑：

基础实现（假设checkDb返回Mono<Optional>）

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;

public class ReactorDemo {
    // 模拟你的IO查询方法，返回Mono包装的Optional
    private Mono<Optional<Object>> checkDb(Integer id) {
        // 替换为实际的DB/IO调用逻辑
        return Mono.just(id == 2 ? Optional.of(new Object()) : Optional.empty());
    }

    public void findFirstValidId() {
        List<Integer> idList = Arrays.asList(1, 2, 3, 4);
        Integer defaultVal = 0;

        Flux.fromIterable(idList)
                // concatMap保证严格按顺序执行IO，前一个完成才发起下一个
                .concatMap(id -> checkDb(id)
                        .map(optRes -> new IdResultPair(id, optRes))
                )
                // 过滤出查询有结果的条目
                .filter(pair -> pair.result().isPresent())
                // 取第一个命中的条目，自动取消上游，不会发起后续IO
                .next()
                // 提取对应id，无命中则返回默认值0
                .map(IdResultPair::id)
                .defaultIfEmpty(defaultVal)
                // 拿到结果后的后续处理逻辑
                .subscribe(firstValidId -> {
                    // 此处编写你原来的后续处理代码
                    System.out.println("第一个有效id为：" + firstValidId);
                });
    }

    // Java 16+可用record，低版本替换为普通POJO类
    private record IdResultPair(Integer id, Optional<Object> result) {}
}

简化实现（如果checkDb用Mono.empty()表示无结果）

如果你的IO方法空结果直接返回空Mono而不是Optional，可以简化写法：

// 模拟IO方法，空结果返回Mono.empty()
private Mono<Object> checkDb(Integer id) {
    return id == 2 ? Mono.just(new Object()) : Mono.empty();
}

public void findFirstValidId() {
    List<Integer> idList = Arrays.asList(1, 2, 3, 4);

    Flux.fromIterable(idList)
            .concatMap(id -> checkDb(id).map(res -> id))
            .next()
            .defaultIfEmpty(0)
            .subscribe(firstValidId -> {
                // 后续处理
            });
}

关键操作符说明 #

• concatMap：替代flatMap，保证元素按原列表顺序依次处理，完全匹配原for循环的执行顺序，不会并发发起IO请求。
• next()：拿到第一个符合条件的元素后立刻向上游发送取消信号，终止后续所有处理，和原代码中的break效果完全一致，不会发起多余的IO请求。
• 如果你需要同步获取结果（非响应式环境调用），可以用block()方法替代subscribe()，注意不要在响应式调度线程中调用block()：

Integer firstValidId = Flux.fromIterable(idList)
        .concatMap(id -> checkDb(id).map(res -> id))
        .next()
        .defaultIfEmpty(0)
        .block();

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