使用ReactiveMongoTemplate实现异步插入同步查询的问题求助

1. 避免在响应式线程中执行阻塞操作 #

你遇到的错误核心是在Reactor的非阻塞IO线程（reactor-http-nio-*）中执行了阻塞迭代操作，这类线程专为非阻塞任务设计，不能被阻塞调用占用。正确做法是用响应式操作链替代阻塞逻辑：

修正异步插入+查询的逻辑 #

如果插入操作无需等待结果、仅需后台异步执行，直接用响应式流处理查询结果即可，无需强行阻塞：

// 异步插入（后台执行，无需等待）
reactiveMongoTemplate.save(myObject).subscribe();

// 非阻塞处理查询结果
reactiveMongoTemplate.find(Query.query(Criteria.where("myId").in(myList)), MyObject.class)
    .doOnNext(myRecord -> {
        // 处理每条查询记录
    })
    .subscribe();

如果需要等待插入完成后再执行查询，用thenMany组合操作：

reactiveMongoTemplate.save(myObject)
    .thenMany(reactiveMongoTemplate.find(Query.query(Criteria.where("myId").in(myList)), MyObject.class))
    .doOnNext(myRecord -> {
        // 处理查询结果
    })
    .subscribe();

修正WebClient中的代码 #

你在flatMap里的阻塞迭代是错误的，需将查询的Flux与repMap的处理整合到响应式流中：

webClient.post()
    .uri("serviceurl")
    .headers(headers -> {
        // 设置请求头
    })
    .body(Mono.just(body), JsonNode.class)
    .retrieve()
    .bodyToMono(MyObject.class)
    .flatMap(responseObject -> {
        Map<String, MySubResponse> repMap = responseObject.stream()
            .collect(Collectors.toMap(MySubResponse::getKey, Function.identity()));
        
        // 用响应式操作处理查询结果，避免阻塞
        return reactiveMongoTemplate.find(Query.query(Criteria.where("myId").in(myList)), MyObject.class)
            .doOnNext(val -> {
                MySubResponse mySubResponse = repMap.get(val.getKey());
                if (mySubResponse != null) {
                    mySubResponse.setMyProperty(val.getProperty());
                }
            })
            // 所有查询处理完成后返回目标结果
            .then(Mono.just(repMap.values()));
    });

2. 混用ReactiveMongoTemplate与MongoTemplate的可行性 #

不推荐直接混用，原因如下：

• 两者依赖完全不同的Mongo客户端：ReactiveMongoTemplate基于异步响应式客户端，MongoTemplate基于同步阻塞客户端，底层连接池、线程模型完全割裂。
• 若在响应式线程中调用MongoTemplate的同步方法，仍会阻塞非阻塞IO线程，引发性能问题甚至报错。

若确实需要同步查询，推荐两种合理方案：

方案一：将同步查询放到专门的阻塞线程池 #

用Mono.fromCallable配合自定义阻塞线程池，避免占用Reactor核心线程：

// 定义用于阻塞操作的线程池
ExecutorService blockingPool = Executors.newFixedThreadPool(10);

// 在WebClient流中使用
webClient.post()
    // ... 省略其他步骤
    .flatMap(responseObject -> {
        Map<String, MySubResponse> repMap = responseObject.stream()
            .collect(Collectors.toMap(MySubResponse::getKey, Function.identity()));
        
        // 同步查询放到阻塞线程池执行
        return Mono.fromCallable(() -> {
            List<MyObject> results = mongoTemplate.find(Query.query(Criteria.where("myId").in(myList)), MyObject.class);
            results.forEach(val -> {
                MySubResponse mySubResponse = repMap.get(val.getKey());
                if (mySubResponse != null) {
                    mySubResponse.setMyProperty(val.getProperty());
                }
            });
            return repMap.values();
        }).subscribeOn(Schedulers.fromExecutor(blockingPool));
    });

方案二：统一使用响应式编程模型 #

尽量将所有操作改为响应式，避免同步阻塞。如果服务基于Spring WebFlux，全响应式模型能最大化发挥非阻塞架构的性能优势。

总结 #

• 严禁在Reactor非阻塞IO线程中执行阻塞操作，始终用响应式操作链替代阻塞逻辑。
• 尽量避免混用ReactiveMongoTemplate与MongoTemplate；若必须用同步操作，务必放到专门的阻塞线程池中执行。
• 基于Spring WebFlux的服务，优先采用全响应式编程模型。

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