Quarkus框架下如何将命令式业务方法改造为响应式实现

问题原因 #

你遇到的是Quarkus响应式事务+Mutiny编程的两个典型错误：

• 第一：@ReactiveTransactional注解的方法不需要你手动调用subscribe()，Quarkus容器会自动订阅你返回的响应式流，事务的开启、提交、回滚完全绑定到流的生命周期上。你手动在方法内subscribe，会导致后续的数据库操作运行在事务已经结束的上下文里，自然不会生效。
• 第二：你在map回调里创建的Multi没有合并到外层的响应式流中，属于孤立的异步操作，既不受事务管理，也没有保证执行完成的机制。同时你直接用map接收persistAndFlush()返回的Uni，得到的是Multi<Uni<Role>>，内层的Uni根本没有被执行。

解决代码 #

首先修改RoleService的seedRoles方法：

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.Multi;

@ReactiveTransactional
public Uni<Void> seedRoles() {
    return Role.existsRoleSeed()
            // 用flatMap承接异步操作的返回值
            .flatMap(exists -> {
                if (exists) {
                    // 已存在数据直接返回完成信号
                    return Uni.createFrom().voidItem();
                }
                // 枚举转Role实例流
                return Multi.createFrom().items(Role.UserRole.values())
                        .map(userRole -> Role.builder().name(userRole).build())
                        // 打平persist返回的Uni，按顺序执行插入
                        .onItem().transformToUniAndConcatenate(role -> role.persist())
                        // 所有插入完成后返回空信号
                        .collect().last()
                        .replaceWithVoid();
            });
}

如果是在启动类调用初始化方法，示例写法如下：

import io.quarkus.runtime.StartupEvent;
import javax.enterprise.event.Observes;

public void onStart(@Observes StartupEvent event) {
    seedRoles().subscribe().with(
        unused -> LOGGER.info("角色初始化完成"),
        failure -> LOGGER.error("角色初始化失败", failure)
    );
}

注意事项 #

• 不需要每个插入都调用flush，事务提交时会自动执行flush操作，减少数据库交互次数。
• 不要在@ReactiveTransactional方法里写阻塞操作，也不要手动管理事务，全部交给容器处理即可。
• 如果是把该方法作为REST接口的实现，直接返回Uni<Void>即可，RESTEasy Reactive会自动处理订阅和响应。

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