WebFlux+JPA应用中，哪层应封装Reactor发布者？

给定场景 #

• 采用阻塞式Spring Data Repository实现数据访问
• 服务层方法标注@Transactional注解，需支持读写事务管理
• Web层基于RouterFunction触发WebFlux处理器处理请求

核心问题 #

确定在哪一层将同步数据操作封装为Reactor发布者（Mono/Flux）

两种分层方案分析 #

选项1：处理器层封装 #

服务层仅提供同步API，负责Repository调用与对象映射；处理器层完成Reactor发布者的封装与线程切换。

处理器代码：

// TaskHandler
@Operation(parameters = @Parameter(in = ParameterIn.PATH, name = "id"), description = "Retrieves task by id.")
@ApiResponse(responseCode = "200", content = @Content(mediaType = "application/json"), description = "Task.")
public Mono<ServerResponse> findById(ServerRequest request) {
    return Mono.fromCallable(() -> UUID.fromString(request.pathVariable("id")))
            .flatMap(id -> Mono.fromCallable(() -> service.findById(id))
                    .subscribeOn(Schedulers.boundedElastic()))
            .flatMap(Mono::justOrEmpty)
            .flatMap(task -> ServerResponse.ok().bodyValue(task))
            .onErrorResume(EntityNotFoundException.class, e -> ServerResponse.status(HttpStatus.NOT_FOUND).bodyValue(e.getMessage()));
}

服务层代码：

// TaskService
@Transactional(readOnly = true)
public Optional<TaskResponseDto> findById(UUID id) {
    return repository.findById(id)
            .map(mapper::toDto)
            .orElseThrow(() -> new EntityNotFoundException("No such task"));
}

优缺点：

• 优点：服务层保持纯同步API，事务上下文不会丢失（整个服务方法调用在boundedElastic线程内执行，ThreadLocal事务上下文有效）；逻辑分工明确。
• 缺点：处理器层需要感知底层阻塞操作的线程切换逻辑，违反关注点分离；服务层无法直接暴露反应式契约。

选项2：服务层封装 #

服务层直接返回Reactor发布者，处理器层仅负责请求响应映射。

处理器代码：

// TaskHandler
@Operation(parameters = @Parameter(in = ParameterIn.PATH, name = "id"), description = "Retrieves task by id.")
@ApiResponse(responseCode = "200", content = @Content(mediaType = "application/json"), description = "Task.")
public Mono<ServerResponse> findById(ServerRequest request) {
    return Mono.fromCallable(() -> UUID.fromString(request.pathVariable("id")))
            .flatMap(service::findById)
            .flatMap(task -> ServerResponse.ok().bodyValue(task))
            .onErrorResume(EntityNotFoundException.class, t -> ServerResponse.status(HttpStatus.NOT_FOUND).bodyValue(t.getMessage()));
}

服务层代码：

// TaskService
@Transactional(readOnly = true)
public Mono<TaskResponseDto> findById(UUID id) {
    return Mono.fromCallable(() -> repository.findById(id))
            .subscribeOn(Schedulers.boundedElastic())
            .flatMap(Mono::justOrEmpty)
            .map(mapper::toDto)
            .switchIfEmpty(Mono.error(new EntityNotFoundException("No such task")));
}

核心缺陷：

使用JpaTransactionManager时，事务上下文存储在ThreadLocal中。当反应式链通过subscribeOn切换到boundedElastic线程池时，原线程的ThreadLocal上下文无法传递，导致Spring无法识别新线程的事务边界：只读方法可能出现懒加载异常，写操作的事务无法正确提交或回滚。

解决方案（不使用R2DBC） #

方案1：优化处理器层封装，保证事务上下文有效 #

保持服务层的同步API与@Transactional注解，在处理器层将整个服务方法调用包裹在Mono.fromCallable中，并通过subscribeOn指定阻塞线程池。这样服务层的事务上下文会在boundedElastic线程内完整生效，避免线程切换导致的上下文丢失。

修改后的处理器代码：

// TaskHandler
public Mono<ServerResponse> findById(ServerRequest request) {
    return Mono.fromCallable(() -> UUID.fromString(request.pathVariable("id")))
            // 将整个服务方法调用放入Callable，确保事务上下文在同一线程内
            .flatMap(id -> Mono.fromCallable(() -> service.findById(id))
                    .subscribeOn(Schedulers.boundedElastic()))
            .flatMap(Mono::justOrEmpty)
            .flatMap(task -> ServerResponse.ok().bodyValue(task))
            .onErrorResume(EntityNotFoundException.class, e -> ServerResponse.status(HttpStatus.NOT_FOUND).bodyValue(e.getMessage()));
}

服务层保持原同步API不变，@Transactional注解正常生效：

// TaskService
@Transactional(readOnly = true)
public Optional<TaskResponseDto> findById(UUID id) {
    return repository.findById(id)
            .map(mapper::toDto)
            .orElseThrow(() -> new EntityNotFoundException("No such task"));
}

方案2：服务层使用TransactionalOperator管理反应式事务 #

如果希望服务层暴露反应式API，可放弃@Transactional注解，改用TransactionalOperator在反应式链中显式管理事务。这样可以手动控制事务边界，避免ThreadLocal上下文丢失问题。

服务层代码修改：

// TaskService
private final TransactionalOperator transactionalOperator;

public Mono<TaskResponseDto> findById(UUID id) {
    return Mono.fromCallable(() -> repository.findById(id))
            .subscribeOn(Schedulers.boundedElastic())
            .flatMap(Mono::justOrEmpty)
            .map(mapper::toDto)
            .switchIfEmpty(Mono.error(new EntityNotFoundException("No such task")))
            // 显式绑定事务
            .as(transactionalOperator::transactional);
}

处理器层代码保持选项2的简洁形式不变。

注意：使用TransactionalOperator时，需要确保配置正确的事务管理器，且所有阻塞操作都在subscribeOn(Schedulers.boundedElastic())中执行，避免阻塞WebFlux的主线程。

总结 #

• 若优先保证事务稳定性与代码兼容性，推荐方案1：服务层保持同步API，处理器层封装反应式逻辑并确保服务方法在阻塞线程池内完整执行。
• 若希望服务层暴露反应式契约，可采用方案2：使用TransactionalOperator显式管理反应式事务，替代@Transactional注解。

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