云原生-Quarkus反应性入门

Posted 2021-05-03 假程序员樱木

Quarkus的多个反应面

Quarkus是反应性的。如果您在引擎盖下看，会发现为Quarkus应用程序提供动力的反应式引擎。该引擎是Eclipse Vert.x（https://vertx.io）。所有网络I/O都通过无阻塞和响应式Vert.x引擎传递。

想象一个传入的HTTP请求。嵌入在Quarkus中的（Vert.x）HTTP服务器接收请求，然后将其路由到应用程序。如果请求的目标是命令式方法（传统的JAX-RS，用@Blocking......注释的代码），则路由层将在工作线程中调用资源方法并在数据可用时写入响应。到目前为止，没有什么新的或杰出的。下图描述了这种行为。在这种情况下，在工作线程上调用应用程序代码，而业务逻辑可以阻止该线程。

但是，如果HTTP请求的目标是响应式方法（使用RESTEasy Reactive，响应式路由的JAX-RS，@Incoming未使用@Blocking...注释的方法），则路由层将在I/O线程上调用路由，从而带来许多好处，例如更高的并发性和表现：

因为Quarkus使用I/O线程来调用的代码，所以节省了上下文切换，避免了大线程池管理，从而提高了资源利用率。但是，代码必须不阻止线程。为什么？因为，I/O线程用于处理多个并发请求。一旦因为需要执行一些I/O而无法处理请求，它就会调度这些I/O并传递一个继续。它释放可以处理另一个请求的线程。计划的I/O完成后，继续执行该操作，返回I/O线程。

因此，许多Quarkus组件在设计时都考虑到了响应性，例如数据库访问（PostgreSQL，mysql，Mongo等），应用程序服务（邮件，模板引擎等），消息传递（Kafka，AMQP等）。等等。但是，要充分利用此模型，应以非阻塞方式编写应用程序代码。那是拥有反应式API的终极武器。

Mutiny-一个反应式编程库

Mutiny是一个反应式编程库，允许表达和编写异步动作。它提供两种类型：

• io.smallrye.mutiny.Uni -用于提供0或1个结果的异步操作

• io.smallrye.mutiny.Multi -用于多物料（带背压）流

两种类型都是惰性的，并且遵循订阅模式。仅在实际需要时（即，订户加入），才开始计算。

uni.subscribe().with(
    result -> System.out.println("result is " + result),
    failure -> failure.printStackTrace()
);

multi.subscribe().with(
    item -> System.out.println("Got " + item),
    failure -> failure.printStackTrace()
);

双方Uni并Multi揭露事件驱动的API：表达想要在一个给定的事件（成功，失败等）做什么。这些API分为几组（操作类型），以使其更具表现力，并且避免将100个方法附加到一个类中。操作的主要类型是对失败，完成，处理项目，提取或收集它们做出反应。它具有可导航的API，可提供流畅的编码体验，并且结果不需要太多有关反应式的知识。

httpCall
    .onFailure().recoverWithItem("my fallback");

反应式流参考（
https://www.reactive-streams.org/）。 Multi实现Reactive Streams Publisher从而实现Reactive Streams背压机制。

拥抱Quarkus的反应性和命令性支柱的统一，Uni并Multi为命令性构造提供桥梁。可以将转换Multi为Iterable或等待产生的项目Uni。

// Block until the result is available
String result = uni.await().indefinitely();

// Transform an asynchronous stream into a blocking iterable
stream.subscribe().asIterable().forEach(s -> System.out.println("Item is " + s));

如果是RxJava或电抗器的用户，可能不知道你如何使用自己熟悉的Flowable，Single，Flux，Mono...兵变允许转换Unis和Multis从和RX Java和反应堆类型：

Maybe<String> maybe = uni.convert().with(UniRxConverters.toMaybe());
Flux<String> flux = multi.convert().with(MultiReactorConverters.toFlux());

但是，Vert.x呢？还可以使用Mutiny类型使用Vert.x API。以下代码片段显示了Vert.x Web客户端的用法：

// Use io.vertx.mutiny.ext.web.client.WebClient
client = WebClient.create(vertx,
                new WebClientOptions().setDefaultHost("fruityvice.com").setDefaultPort(443).setSsl(true)
                        .setTrustAll(true));
// ...
Uni<JsonObject> uni =
    client.get("/api/fruit/" + name)
        .send()
        .onItem().transform(resp -> {
            if (resp.statusCode() == 200) {
                return resp.bodyAsJsonObject();
            } else {
                return new JsonObject()
                        .put("code", resp.statusCode())
                        .put("message", resp.bodyAsString());
            }
        });

最后但并非最不重要的一点是，Mutiny与SmallRye Context Propagation内置了集成，因此可以在反应式管道中传播事务，可追溯性数据等。