Quarkus Mutiny：异步请求的正确处理方式

Quarkus Mutiny：异步请求的正确处理方式

本文旨在帮助开发者理解如何在 Quarkus 中使用 Mutiny 处理异步请求，特别是当需要等待多个请求完成后再进行下一步操作时。我们将通过示例代码展示如何避免阻塞主线程，以及如何正确地组合多个 Uni 对象，确保在所有依赖的异步操作完成后返回最终结果。同时，强调了在 Quarkus 环境下，避免使用 await().indefinetly 和手动 subscribe() 的重要性。

在使用 Quarkus 和 Mutiny 构建响应式应用时，一个常见的挑战是如何处理多个异步操作，并确保在所有操作完成后再返回结果。例如，你需要调用多个外部 API，并将它们的结果组合起来。直接使用 await().indefinitely 可能会阻塞 Vert.x 的事件循环线程，导致性能问题。手动 subscribe() 在Quarkus中也通常是不需要的，因为Quarkus能够自动处理异步结果。

下面，我们将通过示例代码演示如何正确地处理这种情况。

使用 Uni.combine().all() 组合多个 Uni 对象

假设我们需要调用三个不同的 API 来获取汽车的门、轮子和窗户信息，然后将这些信息组合成一个 Car 对象。

首先，定义获取这些信息的三个方法：

Baklib

在线创建产品手册、知识库、帮助文档

下载

Uni> getDoors(String variable1, String variable2, String variable3) {
    // 模拟API调用，返回门的列表
    return Uni.createFrom().item(List.of(new JsonObjectCar("door1"), new JsonObjectCar("door2")))
            .onItem().delayIt().by(Duration.ofMillis(500)); // 模拟延迟
}

Uni> getWheels(String variable1, String variable2, String variable3) {
    // 模拟API调用，返回轮子的列表
    return Uni.createFrom().item(List.of(new JsonObjectCar("wheel1"), new JsonObjectCar("wheel2")))
            .onItem().delayIt().by(Duration.ofMillis(300)); // 模拟延迟
}

Uni> getWindows(String variable1, String variable2, String variable3) {
    // 模拟API调用，返回窗户的列表
    return Uni.createFrom().item(List.of(new JsonObjectCar("window1"), new JsonObjectCar("window2")))
            .onItem().delayIt().by(Duration.ofMillis(700)); // 模拟延迟
}

class JsonObjectCar {
    private String name;

    public JsonObjectCar(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }
}

class Car {
    private List doors;
    private List wheels;
    private List windows;

    public void setDoors(List doors) {
        this.doors = doors;
    }

    public void setWheels(List wheels) {
        this.wheels = wheels;
    }

    public void setWindows(List windows) {
        this.windows = windows;
    }

    public List getDoors() {
        return doors;
    }

    public List getWheels() {
        return wheels;
    }

    public List getWindows() {
        return windows;
    }
}

接下来，使用 Uni.combine().all() 将这三个 Uni 对象组合起来：

import io.smallrye.mutiny.Uni;
import java.time.Duration;
import java.util.List;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import java.util.Optional;

@Path("/api")
public class CarResource {

    @GET
    @Path("/testingAsync")
    public Uni testingMutiny() {
        String variable1 = "var1";
        String variable2 = "var2";
        String variable3 = "var3";

        Uni> carDoorsUni = getDoors(variable1, variable2, variable3);
        Uni> carWheelsUni = getWheels(variable1, variable2, variable3);
        Uni> carWindowsUni = getWindows(variable1, variable2, variable3);

        return Uni.combine()
                .all()
                .unis(carDoorsUni, carWheelsUni, carWindowsUni)
                .combinedWith(list -> {
                    // Result of carDoorsUni
                    List carDoors = list.get(0);

                    // Result of carWheelsUni
                    List carWheels = list.get(1);

                    // Result of carWindowsUni
                    List carWindows = list.get(2);

                    // Create a car instance with the previous results
                    Car car = new Car();
                    Optional.ofNullable(carDoors).ifPresent(car::setDoors);
                    Optional.ofNullable(carWheels).ifPresent(car::setWheels);
                    Optional.ofNullable(carWindows).ifPresent(car::setWindows);

                    // You can also return a list of cars, but you need to change the return type of testingMutiny to Uni>
                    return car;
                })
                .invoke(() -> System.out.println("Okay it worked"));
    }

    Uni> getDoors(String variable1, String variable2, String variable3) {
        // 模拟API调用，返回门的列表
        return Uni.createFrom().item(List.of(new JsonObjectCar("door1"), new JsonObjectCar("door2")))
                .onItem().delayIt().by(Duration.ofMillis(500)); // 模拟延迟
    }

    Uni> getWheels(String variable1, String variable2, String variable3) {
        // 模拟API调用，返回轮子的列表
        return Uni.createFrom().item(List.of(new JsonObjectCar("wheel1"), new JsonObjectCar("wheel2")))
                .onItem().delayIt().by(Duration.ofMillis(300)); // 模拟延迟
    }

    Uni> getWindows(String variable1, String variable2, String variable3) {
        // 模拟API调用，返回窗户的列表
        return Uni.createFrom().item(List.of(new JsonObjectCar("window1"), new JsonObjectCar("window2")))
                .onItem().delayIt().by(Duration.ofMillis(700)); // 模拟延迟
    }

    class JsonObjectCar {
        private String name;

        public JsonObjectCar(String name) {
            this.name = name;
        }

        public String getName() {
            return name;
        }
    }

    class Car {
        private List doors;
        private List wheels;
        private List windows;

        public void setDoors(List doors) {
            this.doors = doors;
        }

        public void setWheels(List wheels) {
            this.wheels = wheels;
        }

        public void setWindows(List windows) {
            this.windows = windows;
        }

        public List getDoors() {
            return doors;
        }

        public List getWheels() {
            return wheels;
        }

        public List getWindows() {
            return windows;
        }
    }
}

在这个例子中，Uni.combine().all().unis() 接收一个包含多个 Uni 对象的列表，并返回一个新的 Uni 对象，该对象在所有输入的 Uni 对象都发出值后才发出一个值。combinedWith 方法接收一个函数，该函数将所有输入 Uni 对象发出的值组合成一个结果。

注意事项:

• 避免阻塞: 不要在 Vert.x 事件循环线程中调用 await().indefinitely 或其他阻塞方法。这会降低应用的响应速度和吞吐量。
• 错误处理: 如果任何一个 Uni 对象发出错误，组合后的 Uni 对象也会发出错误。你可以使用 onFailure() 方法来处理错误。
• Quarkus 的自动处理: 在 Quarkus 中，通常不需要手动 subscribe() Uni 或 Multi 对象。Quarkus 会自动处理异步结果。
• 线程模型： 了解 Vert.x 的线程模型至关重要。避免在事件循环线程中执行耗时操作。将这些操作 offload 到 worker 线程。

总结

使用 Quarkus Mutiny 处理异步请求的关键在于理解响应式编程的思想，并利用 Mutiny 提供的组合操作符来处理多个异步操作。避免阻塞事件循环线程，并利用 Quarkus 的自动处理能力，可以构建高效、响应迅速的应用程序。Uni.combine().all() 是一个强大的工具，可以帮助你轻松地组合多个 Uni 对象，并在所有操作完成后执行下一步操作。