Mongo Reactive Repository在Filter中调用阻塞方法致挂起求助

问题根源 #

你遇到的是响应式流中阻塞操作引发的线程死锁问题。

在Reactive编程模型里，Mongo Reactive Repository的操作运行在Reactor的I/O线程池中，当你在filter操作符里调用blockOptional()时，会直接阻塞当前的Reactive线程——而这个线程正是要执行incotermLocationRepository数据库查询的线程，最终形成死锁：

1. 初始代码中，deliveryRepository.findById()返回的Mono在Reactive线程上执行，进入第一个filter后调用deliveryStatusShouldBeUpdatedToDelivered，里面的blockOptional()会强制等待异步任务完成。
2. 但incotermLocationRepository的查询需要从同一个Reactive线程池获取线程执行，当前线程已被阻塞，任务无法调度，导致Repository调用一直挂起。

修改后的代码把blockOptional()移到了deliveryRepository.findById()之后，阻塞的是外层forEach的普通线程（非Reactive I/O线程），后续调用incotermLocationService时，Reactive线程能正常调度，死锁问题自然解决。

正确的响应式写法（避免阻塞） #

既然用了Reactive，就要全程保持非阻塞，不要在流中间调用block()/blockOptional()。正确做法是把业务逻辑改成响应式，用filterWhen替代filter处理异步判断：

重构代码示例 #

shipment.getItems().forEach(item -> {
    deliveryRepository.findById(item.getDeliveryNumber())
        .filter(delivery -> Objects.isNull(delivery.getState()) || !delivery.getState().equals(TrackObjectState.DELIVERED))
        // 用filterWhen处理异步过滤逻辑
        .filterWhen(delivery -> shouldUpdateToDeliveredReactive(delivery, shipment, gamMessage))
        .flatMap(delivery -> {
            delivery.setState(TrackObjectState.DELIVERED);
            return deliveryRepository.save(delivery);
        })
        .subscribe(delivery -> log.info("delivery: " + delivery.getDeliveryNumber() + " is updated to state delivered"));
});

// 重构业务方法为响应式
private Mono<Boolean> shouldUpdateToDeliveredReactive(Delivery delivery, Shipment shipment, GamMessage gamMessage) {
    String destinationLocation = ...; // 原逻辑中获取目标位置的代码
    return incotermLocationService.getAlternativeUnLocationCodesByUnlocationCode(destinationLocation)
        .map(alternativeLocations -> {
            // 原deliveryStatusShouldBeUpdatedToDelivered中的判断逻辑
            return /* 这里返回原方法的布尔判断结果 */;
        });
}

关键说明 #

• 用filterWhen替代filter：filterWhen接受返回Mono<Boolean>的函数，允许在过滤逻辑中执行异步操作，全程保持非阻塞。
• 移除所有阻塞调用：让整个流保持响应式链式调用，线程能正常调度，避免死锁。
• 业务方法响应式改造：把包含阻塞调用的业务方法改成返回Mono/Flux，确保流程全程非阻塞。

总结 #

在Reactive编程中，绝对不要在响应式流的操作符（如filter、map、flatMap）内部调用阻塞方法，这会破坏Reactive线程模型，导致线程死锁或资源耗尽。正确做法是全程使用响应式API，用Reactor提供的操作符处理异步逻辑，保持流的非阻塞特性。

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