Quarkus Mutiny使用Unis.combinedWith时请求被忽略问题求助

问题描述 #

我创建了三个分别调用不同微服务的Uni对象：

Uni<List<JsonObjectCar>> carDoorsUni = getDoors(variable1, variable2, variable3);

Uni<List<JsonObjectCar>> carWheelsUni = getWheels(variable1, variable2, variable3);

Uni<List<JsonObjectCar>> carWindowsUni = getWindows(variable1, variable2, variable3);

尝试用以下代码合并响应，但IntelliJ提示合并操作被忽略，实际从未执行：

Uni.combine()
        .all()
        .unis(carDoorsUni, carWheelsUni, carWindowsUni)
        .combinedWith((carDoors, carWheels, carWindows) -> {
            Optional.of(carDoors)
                    .ifPresent(val -> car.setDoors(val.getDoors()));
            Optional.of(carWheels)
                    .ifPresent(val -> car.setWheels(val.getWheels()));
            Optional.of(carWindows)
                    .ifPresent(val -> car.setWindows(val.getWindows()));

            return car;
        });

添加.await().indefinitely()后又收到错误：The current thread cannot be blocked: vert.x-eventloop-thread-11。

核心原因与解决方法 #

1. Uni是惰性执行的，必须订阅才会触发 #

Quarkus/Vert.x中的Uni遵循响应式流规范，仅构建操作链不会触发执行，必须通过订阅方法（如subscribe().with(...)）或交由框架处理（比如作为REST接口返回值），整个链式操作才会被执行。之前的代码只定义了合并逻辑，但没有触发执行，因此被忽略。

2. 禁止在EventLoop线程阻塞 #

Vert.x的EventLoop线程是IO线程，要求全程非阻塞，调用await()会阻塞线程，违反Vert.x线程模型，导致报错。必须保持异步流程，通过订阅处理结果。

修正后的代码示例 #

普通异步场景 #

// 构建合并操作链
Uni<Car> combinedUni = Uni.combine()
        .all()
        .unis(carDoorsUni, carWheelsUni, carWindowsUni)
        .combinedWith((carDoors, carWheels, carWindows) -> {
            // 建议用ofNullable避免null触发NoSuchElementException
            Optional.ofNullable(carDoors)
                    .ifPresent(val -> car.setDoors(val.getDoors()));
            Optional.ofNullable(carWheels)
                    .ifPresent(val -> car.setWheels(val.getWheels()));
            Optional.ofNullable(carWindows)
                    .ifPresent(val -> car.setWindows(val.getWindows()));

            return car;
        });

// 订阅触发执行，处理结果和异常
combinedUni.subscribe().with(
        updatedCar -> {
            // 处理合并完成后的Car对象，比如写入数据库、调用下游服务等
        },
        throwable -> {
            // 处理合并过程中出现的异常，比如日志记录、错误返回等
        }
);

REST接口场景 #

如果代码位于JAX-RS/RESTEasy端点中，直接返回合并后的Uni即可，框架会自动处理订阅和线程调度：

@GET
@Path("/combined-car")
public Uni<Car> getCombinedCar() {
    Uni<List<JsonObjectCar>> carDoorsUni = getDoors(variable1, variable2, variable3);
    Uni<List<JsonObjectCar>> carWheelsUni = getWheels(variable1, variable2, variable3);
    Uni<List<JsonObjectCar>> carWindowsUni = getWindows(variable1, variable2, variable3);

    // 初始化Car对象，根据实际业务调整
    Car car = new Car();

    return Uni.combine()
            .all()
            .unis(carDoorsUni, carWheelsUni, carWindowsUni)
            .combinedWith((carDoors, carWheels, carWindows) -> {
                Optional.ofNullable(carDoors)
                        .ifPresent(val -> car.setDoors(val.getDoors()));
                Optional.ofNullable(carWheels)
                        .ifPresent(val -> car.setWheels(val.getWheels()));
                Optional.ofNullable(carWindows)
                        .ifPresent(val -> car.setWindows(val.getWindows()));

                return car;
            });
}

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