如何将Mutiny的combine().all()改为同步链式调用适配Hibernate Reactive限制

我使用Quarkus 3.13.0搭配Hibernate Reactive开发，原本通过Mutiny的Uni.combine()并行查询两个实体，再用查询结果持久化关联表，但遇到Hibernate的限制——同一Session不允许并行执行操作，触发如下错误：

2024-08-06 13:20:30,632 ERROR [http-problem] (vert.x-eventloop-thread-1) status=500, title="Internal Server Error": io.smallrye.mutiny.CompositeException: Multiple exceptions caught:
        [Exception 0] java.lang.IllegalStateException: Session/EntityManager is closed
        [Exception 1] io.vertx.core.impl.NoStackTraceThrowable: Transaction already complete
        at io.smallrye.mutiny.groups.UniOnFailure.lambda$call$3(UniOnFailure.java:108)

刚接触Mutiny，找不到combine()之外带BiFunction的替代方式，希望把以下代码改成同步链式调用以解决错误：

@POST
@Path("{userId}/teams/{businessUnitId}")
public Uni<RestResponse<UserDto>> assignToTeam(Long userId, Long businessUnitId, @RestQuery String pRole) {
    if (businessUnitId == null || userId == null) {
        throw new WebApplicationException(MSG_INVALID_PAYLOAD, 422);
    }

    return Uni.combine()
            .all()
            .unis(Panache.withTransaction(() -> userRepo.findById(userId)
                    .onItem()
                    .ifNull()
                    .failWith(new WebApplicationException(MSG_RESOURCE_NOT_FOUND, NOT_FOUND))),
                    Panache.withTransaction(() -> businessUnitRepo.findById(businessUnitId)
                            .onItem()
                            .ifNull()
                            .failWith(new WebApplicationException(MSG_RESOURCE_NOT_FOUND, NOT_FOUND))))
            .with((user, bunit) -> {
                BusinessUnitsUsers buu = new BusinessUnitsUsersBuilder().parentEntity(bunit)
                        .childEntity(user)
                        .role(pRole)
                        .build();
                return user.assingToTeam(buu);
            })
            .chain(r -> Panache.withTransaction(() -> userRepo.persist(r))
                    .map(user -> RestResponse.ok(userMapper.toDto(user, usersBusinessUnitContext))));
}

修改后的链式调用代码 #

@POST
@Path("{userId}/teams/{businessUnitId}")
public Uni<RestResponse<UserDto>> assignToTeam(Long userId, Long businessUnitId, @RestQuery String pRole) {
    if (businessUnitId == null || userId == null) {
        throw new WebApplicationException(MSG_INVALID_PAYLOAD, 422);
    }

    // 第一步：查询User，不存在则抛出404
    return Panache.withTransaction(() -> userRepo.findById(userId)
                    .onItem()
                    .ifNull()
                    .failWith(new WebApplicationException(MSG_RESOURCE_NOT_FOUND, NOT_FOUND)))
            // 第二步：拿到User后，查询对应的BusinessUnit，不存在则抛出404
            .chain(user -> Panache.withTransaction(() -> businessUnitRepo.findById(businessUnitId)
                    .onItem()
                    .ifNull()
                    .failWith(new WebApplicationException(MSG_RESOURCE_NOT_FOUND, NOT_FOUND))
                    // 将User和BusinessUnit配对，传给下一步
                    .map(bunit -> Pair.of(user, bunit))))
            // 第三步：创建关联实体，更新User的关联关系
            .chain(pair -> {
                User user = pair.getLeft();
                BusinessUnit bunit = pair.getRight();
                BusinessUnitsUsers buu = new BusinessUnitsUsersBuilder()
                        .parentEntity(bunit)
                        .childEntity(user)
                        .role(pRole)
                        .build();
                User updatedUser = user.assingToTeam(buu);
                // 持久化更新后的User
                return Panache.withTransaction(() -> userRepo.persist(updatedUser));
            })
            // 最后：将持久化后的User转为DTO，返回成功响应
            .map(user -> RestResponse.ok(userMapper.toDto(user, usersBusinessUnitContext)));
}

代码说明 #

1. 链式调用逻辑：用chain()替代并行组合，保证每个数据库操作按顺序执行，共享同一个Session/事务，规避Hibernate的并行操作限制。
2. Pair传递参数：链式调用仅能传递单个结果，用Pair（或自定义DTO）将User和BusinessUnit的结果配对，传给后续步骤处理关联关系。
3. 事务边界：每个数据库操作都包裹在Panache.withTransaction()中，确保操作在事务内执行，避免Session提前关闭的问题。

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