拥抱 JVM 上的反应式应用：深入理解现代 I/O 模型和 Vert.x

概述

本文探讨了从阻塞式 I/O 转向非阻塞式 I/O 和异步 I/O 的过程，强调了它们在现代软件开发中的重要性。文章强调了云计算、大数据和 IoT 驱动的反应式应用程序的兴起。具体来讲，文章主要关注 Vert.x，这是一个在 JVM 上构建反应式应用程序的强大工具包，其知名的特性包括 Multi-Reactor 模式、事件总线和 Verticle。基准测试结果表明，Vert.x 性能卓越，是高并发环境的理想选择。真实使用案例和 COVID-19 全球仪表盘的案例研究说明了 Vert.x 的实际优势和应用。

探索 I/O 模型的演进

• BIO（阻塞式 I/O）：这种模型是同步和阻塞的。由专门的线程处理每个客户端连接，这会导致资源使用效率的低下，尤其是在高负载的情况下。

• NIO（非阻塞式 I/O）：NIO 同步运行，但不阻塞。服务器上的单个线程可同时管理多个客户端请求，这提升了可扩展性。NIO 架构包含三个主要的组件：Buffer、Channel 和 Selector。

• AIO（异步 I/O 或 NIO 2.0）：AIO 通过引入一个异步、非阻塞的模型扩展了 NIO，该模型使用回调来处理相关的操作。这种模型非常适合需要维持大量连接或者需要较长连接持续时间的应用程序。

企业级软件在 I/O 方面的趋势

最近，应用程序的需求发生了明显的变化，尤其是伴随着 云计算、大数据 和 IoT 的兴起。要在现代应用程序领域存活下来，采用反应式架构风格是至关重要的。Play 和 Vert.x 等框架的流行彰显了对非阻塞 I/O 日益增长的需求。Play 利用 Akka 和非阻塞 I/O，集成了 Netty 和 Akka HTTP 以优化操作。Vert.x 的设计是事件驱动和非阻塞的，使其能够以较少的线程高效支持数量众多的并发用户。按照 TechEmpower 的 Web 框架基准测试，使用 Vert.x 开发的应用程序在各种标准测试中都超过了其竞争对手。这些技术的典型实现包括 MasterCard 对 Vert.x 的使用 以及 Netflix 的网关转向异步 NIO。但是，很多像 Capital One 和 Red Hat 这样的许多组织也在其技术栈中使用了 Vert.x。

例如，MasterCard 之所以选择使用 Vert.x 作为支付网关就在于其事件驱动、非阻塞的架构，这确保了高吞吐量和低延迟。Vert.x 处理大量并发事务的能力对于每天处理数亿笔支付的场景至关重要。该框架的横向和纵向可扩展性使 Mastercard 能够满足不断增长的需求。除此之外，Vert.x 工作者（worker）verticle 能够管理长时间运行的任务，而不会阻塞主事件循环，从而保持系统的响应速度。这使得 Vert.x 成为构建有韧性、高性能和高可用性支付系统的理想方案。

另外一个有趣的使用场景是 Dream11，它需要一个高性能、可靠的框架来处理体育比赛期间难以预测的流量激增。单场比赛的观众人数可能会达到 5 亿。粉丝们登录 Dream 11 创建梦之队、观看比赛状态、进行交流等等，这都给扩展性带来了严峻的挑战。在对现有的框架进行评估之后，Dream11 发现它还需要满足性能、函数式编程和 ZIO 支持等要求。Vert.x 提供了非常好的性能，但是缺少函数式的 Scala 特性。由于时间紧迫，Dream11 利用自己在可扩展性系统方面的专业技能，在公司内部构建了 ZIO HTTP。Dream11 对 ZIO HTTP 进行了广泛地优化，实现了超越 Vert.x 和其他框架的性能。现在，ZIO HTTP 能够高效处理数百万个并发请求，成为 Dream11 为 Scala 开发的性能和功能最强的 HTTP 库。

• Vertx-rest：对 resteasy-vertx 的抽象，简化基于 JAX-RS 注解的 vert.x REST 应用程序的编写。

• AeroSpike Client：Vert.x Aerospik 客户端，提供了与 Aerospike 服务器交互的异步 API。

体育比赛都是有季节性的。具体的比赛还取决于参赛队伍、观众以及它们的号召力。Dream11 的用户数量超过了 1 亿。我们看一些使用场景，以更好地理解 Vert.x 是如何满足这些需求的。

板球比赛期间 Dream11 流量的弹性特征

样例 1：赛前激增

在大型的板球比赛开始前，数百万的用户涌入 Dream11 来创建他们的梦之队。这会导致流量在短时间内激增。在此期间，用户会：

• 登录账户

• 浏览球员的统计数据

• 确定他们的梦之队

• 在比赛前做最后的更改

样例 2：赛中参与

在比赛期间，平台的用户流量依然会比较高：

• 查看实时的比分和球员表现

• 参与论坛和社交功能的讨论

• 如果平台允许赛中更改或更新，那么用户可以做出实时决策

样例 3：赛后分析

比赛结束后，用户会返回平台：

• 回顾自己球队的表现

• 查看排名和积分

• 提取奖金或为未来的比赛组件球队

如果想要了解更多详细信息，请参阅 Hackerrank 博客上对 Dream11 首席技术官 Amit Sharma 的访谈全文。

“我们在编程语言和框架方面做了一些标准化工作——90% 以上的服务都是使用 vert.x 编写的。”

——Dream11 首席技术官 Amit Sharma

我们深入探讨一下 Vert.x 为何适用于大型分布式应用程序的构建，然后介绍一个基准测试和参考案例的研究。

理解 Reactor 模型

Reactor 模型是事件驱动架构的基石。它的特点是单线程事件循环。在该模型中，事件（包括像数据库就绪或缓冲区完成这样的 I/O 事件）在发生时会进行排队。例如，一个轻量级的高性能 web 应用需要处理数千个并发连接。服务器需要管理客户端请求，包括 HTTP 请求、数据库查询和文件 I/O 操作。传统的多线程方式是资源密集型的，每个客户端请求都可能产生一个新的线程，最终会导致很高的内存占用率和上下文切换开销，尤其是在面临高负载的情况下。但是，Reactor 模型使用事件循环来处理连接，不会阻塞 I/O。

Reactor：该组件在一个称为事件循环的专属线程上运行，它能够高效地将传入的 I/O 事件路由到指定的处理程序。它可以自主运行，确保无缝的事件处理与分发。

处理程序（Handler）：Reactor 负责协调 I/O 事件，指导处理程序在不阻塞的情况下处理特定的任务。这种方式利用了 NIO 的核心优势，支持非阻塞通信和事件驱动处理。处理程序管理未处理的 I/O 活动，并快速高效地执行操作。这个模型的基础是 NIO 的强大功能，确保系统性能流畅和响应迅速。

反应式系统的四个属性

正如反应式宣言（Reactive Manifesto） 所述，反应式系统的设计目标是：

• 响应性（Responsive）：系统始终能够及时响应，确保稳定的用户体验。

• 韧性（Resilient）：即便在遇到故障时，系统也能保持响应性。这种韧性是通过副本、封闭、隔离和委托来实现的。

• 弹性（Elastic）：系统可以高效地调整其资源，以适应不同的负载，确保在不同的运行条件下具备一致的性能表现。

• 消息驱动（Message-Driven）：系统依赖于异步的消息传递，以确保松耦合、隔离和位置透明。这种方式也增强了负载管理和错误处理能力。

图片来源

使用 Eclipse Vert.x 在 JVM 上构建反应式应用

"Eclipse Vert.x 是一个在 JVM 上构建反应式应用的工具包。"我们需要将 Vert.x 视为一个工具包，而不是框架。这种差异凸显了其灵活性和模块化的特点，允许开发人员选择应用程序所需的组件。

Vert.x 使用 Netty 项目构建，该项目以其在 JVM 上的高性能和异步网络功能而闻名。这个坚实的基础使得 Vert.x 成为开发高效、可扩展和非阻塞 I/O 操作的反应式应用程序的理想选择。

Vert.x 的主要特性（核心）

• Multi-Reactor 模式增强了 Vert.x 中处理并发操作的可扩展性和效率。与使用单事件循环的传统模型不同，Vert.x 使用了多事件循环，通常机器上每个可用的 CPU 核心对应一个事件循环，但这个数量也可以进行手动调整。

• 事件总线是 Vert.x 应用程序中的通信基石，可以实现应用程序不同组成部分之间的各种通信模式。它支持：

• 点对点消息：双方之间的直接通信。

• 请求 - 响应消息：一种双向通信模式，在请求之后返回响应。

• 发布 / 订阅：允许向多个订阅者广播消息。

• Verticle 是 Vert.x 中代码执行的基本单位，封装了 Vert.x 部署和管理的逻辑。Vertice 运行在事件循环之上来处理事件。比如，网络数据、定时器事件或 vertical 之间的消息。Verticle 主要有两种类型：

• 标准 Vertice：始终运行在事件循环线程上，确保非阻塞操作。

• 工作者 Verticle：在工作者池的线程上运行，这个池是为可能阻塞事件循环的任务专门设计的。

• executeBlocking()方法允许执行阻塞式的代码。它使用工作者池中的线程来运行特定的处理程序，因此，能够防止主事件循环被阻塞，这确保了应用程序的响应速度和性能。

比较分析

• 性能：Vert.x 和 Akka 都是为高性能、非阻塞操作而设计的，能够在多个核心之间良好扩展，而 Jetty，尽管也很高效，但是由于其同步的特征，传统上处理的并发连接数较少。

• 复杂性：Jetty 易于使用，可满足简单 Web 服务器的要求；Vert.x 为反应式应用提供了一个灵活的工具包，不过它更为复杂一些；Akka 提供了强大的抽象，但在掌握 Actor 模型方面的学习曲线较为陡峭。

• 适用性：对于传统的 web 应用以及需要嵌入 HTTP/servlet 功能时，可以选择 Jetty；Vert.x 非常适合 JVM 上的异步、反应式应用程序；Akka 最适合需要健壮性、优秀并发性和基于 Actor 编排的系统。

使用 Vert.x 创建反应式 HTTP 服务器

如下是一个使用 Vert.x 在 Java 中创建 HTTP 服务器的样例。服务器监听 8080 端口，并对每个请求均响应一条简单的消息。

import io.vertx.core.Vertx;import io.vertx.core.http.HttpServer;

public class SimpleHttpServer {    public static void main(String[] args) {        Vertx vertx = Vertx.vertx();❶        HttpServer server = vertx.createHttpServer();❷        server.requestHandler(request -> {❸            request.response()                   .putHeader("content-type", "text/plain")                   .end("Hello from Vert.x HTTP Server!");        });        server.listen(8080, result -> {❹            if (result.succeeded()) {                System.out.println("Server is now listening on port 8080!");            } else {                System.out.println("Failed to bind to port 8080: " + result.cause().getMessage());            }        });    }}

❶ 创建 Vertx 实例：本行代码初始化一个新的 Vertx 实例。Vert.x 提供了低层级的 API 和构建块。很多 Vert.x 扩展均使用 Vert.x 核心。

❷ 创建 HTTP 服务器：本行代码创建一个新的 HTTP 服务器实例。

❸ 处理传入的请求：这个代码块创建一个请求处理程序，并使用一个简单的文本消息“Hello from Vert.x HTTP Server!”来响应每个传入的 HTTP 请求。

❹ 监听 8080 端口：这个代码块启动服务器并将其绑定到 8080 端口。如果服务器启动成功，它会打印一条成功的消息。如果失败的话，则会打印一条错误信息，并说明失败原因。

基准测试详情

随着 web 应用程序的复杂性和用户群的增加，web 服务器技术的选择变得至关重要。不同的 web 服务器在处理并发连接、快速处理请求和有效管理资源方面的能力各不相同。

在实际负载的条件下对不同的 web 服务器的性能进行基准测试至关重要。本节比较了两种广泛使用的 web 服务器框架的吞吐量性能，即 Jetty9 和 Vert.x。

所用工具:

WRK：一款现代化的基准测试工具，在单个多核 CPU 上运行时能够生成大量的负载。

硬件规格：

处理器名称：Quad-Core Intel Core i7

处理器速度：2.3 GHz

处理器数量：1

核心总数：4

超线程技术：已启用

内存：32 GB

软件规格：

操作系统：macOS Ventura 13.0

JDK：OpenJDK 17

基准测试的目的

本次基准测试的首要目标是评估和比较两个主要 web 服务器的吞吐量：

1. Jetty9 HTTP Web Server

2. Vert.x HTTP Web Server

吞吐量是一个关键的度量指标，表明一个 web 服务器每秒钟可以处理多少个请求。吞吐量越高，就意味着性能越好，就能同时为更多的用户提供服务。

基准测试的命令：

wrk -t10 -c100 -d30s http://127.0.0.1:8080/v1/ -s post.lua

• -t10：使用的线程数（10 个线程）。

• -c100：保持打开的连接数（100 个连接）

• -d30s：测试的持续时间（30 秒）

• -http://127.0.0.1:8080/v1/：被测服务的 URL

• -s post.lua：用于更复杂请求场景的 Lua 脚本（如果需要的话）

要在本地重现测试结果的话，请检出该 git 仓库。

结果

结论

根据基准测试的结果，与 Jetty9 HTTP Web 相比，Vert.x HTTP Web Server 是高吞吐应用程序的更好选择。它每秒可处理的请求数更高，因此适合需要高性能和高可扩展性的应用程序。

这些结果为高性能应用程序在选择 Web 服务器时做出明智决策提供了有价值的信息。

案例研究：为 COVID-19 全球仪表盘构建 MultiReactor WebApp

概览：在本案例研究中，我们将会创建一个 Web 应用程序，使用 CollectAPI 服务按照国家和全球展示 COVID-19 的统计数据和最新的新冠病毒新闻。我们采用的方式包括维护定期刷新的本地缓存，以存储统计数据，同时始终将对最新新闻的请求直接委托给CollectAPI。

策略和实现

我们将使用 Maven 构建一个 Spring Boot 应用程序，并使用 Vert.x（NIO）工具集实现高吞吐，它不会因为阻塞操作而产生不必要的开销。核心应用程序将部署一个初始的 Vert.x verticle，作为 HTTP web 服务器，并监听 8080 端口。主（primary）verticle 处理所有的 REST API 请求，并根据 Vert.x 路由器中定义的路由对其进行处理。

辅助（secondary）verticle 接收来自主 verticle 的所有请求，并担当请求转发器，将请求路由至相应的服务处理程序。我们将实现两个额外的 verticle 来处理这些请求：

1. Redis Manager：该 verticle 将会管理仪表盘查询请求，并定期刷新缓存数据。

2. CollectAPI Manager：该 verticle 将作为代理，将请求委托给后端 CollectAPI 服务。

所有的 verticle 将通过 Vert.x 事件总线以事件的方式发送和接收请求 / 响应，从而确保 verticle 之间的高效通信，并确保可扩展的无阻塞架构。

时序图

MainApplication

入口点负责搭建 Spring Boot 应用程序、初始化 Vert.x 实例并部署名为CovidDashboardService的 verticle。

@SpringBootApplication@ComponentScan("server")public class MainApplication {

    @Autowired    private CovidDashboardService serverVerticle;

    public static void main(String[] args) {        SpringApplication.run(MainApplication.class, args); ❶    }

    @PostConstruct    public void deployVerticles() {        Vertx vertx = Vertx.vertx(); ❷        vertx.deployVerticle(serverVerticle, response -> {❸           // Log response status         });    }}

❶ 运行 Spring Boot 应用：本行代码启动 Spring Boot 应用。❷ 创建 Vertx 实例：本行代码初始化一个新的 Vertx 实例，即 Vert.x 的核心。

❸ 部署第一个 Verticle：该代码块会部署CovidDashboardService Verticle。如果部署成功的话，则打印成功消息。如果失败的话，则会打印一条包含失败原因的错误消息。

CovidDashboardService

这个类使用特定的路由和处理程序设置 HTTP 服务器。它集成了 Vert.x 事件总线，可以处理缓存的 GET 请求并部署额外的工作者变量。

@Component@NoArgsConstructor

public class CovidDashboardService extends AbstractVerticle {   public CovidDashboardService(Vertx vertx) {        this.vertx = vertx;    }    @Override    public void start() throws Exception {        super.start();        final Router router = Router.router(vertx); ❶        router.route().handler(BodyHandler.create());                   

        addCachedGETRoute(router, List["/cache1","/cache2","cache3"])   ❷         addProxyGETRoute(router, List["/proxy1","/proxy2","proxy3"]) ❸

        HttpServer httpServer = vertx.createHttpServer()  ❹   .requestHandler(router)                .exceptionHandler(exception -> {//Log exception })                .connectionHandler(res -> {//Log new connection });         httpServer.listen(port);

    // deploy next verticle RequestRouteDispatcher         vertx.deployVerticle(RequestRouteDispatcher::new,  ❺        new DeploymentOptions().setWorker(true).setWorkerPoolName("API-Router")        .setWorkerPoolSize(20),result -> { //Log result status });   }private void addCachedGETRoute(Router router, List pathList) {❻        Arrays.stream(pathList).iterator() .forEachRemaining(apiPath -> router.get(apiPath.value)        .handler(this::handleCachedGetRequest));}private void handleCachedGetRequest(RoutingContext ctx) {❼      vertx.eventBus().request("CACHED", ctx.normalizedPath(), response -> {            if (response.succeeded()) { // ctx.response().end("{result}")}            else { // ctx.setStatusCode(500))}      });}}

❶ 路由器初始化：本行代码初始化一个新的 Router 实例。

❷ 添加缓存路由：将缓存查询的所有 API 路径添加到 Router。

❸添加代理路由：将代理查询的所有 API 路径添加到 Router。

❹ 创建并启动 HTTP 服务器：该代码块启动服务器并将其绑定到 8080 端口。

❺ 部署 RequestRouteDispatcher Verticle：用来进行请求派发的下一个 vertical。

❻ 添加和处理缓存的 GET 路由：该方法将缓存查询的 GET 路径添加到路由器中。

❼ 添加和处理代理的 GET 路由：该方法将代理查询的 GET 路径添加到路由器中。

RequestRouteDispatcher

这个类为缓存和代理相关的事件总线设置了消费者，并部署了用来管理 Redis 和数据收集功能的额外 verticle。每个消费者负责处理特定类型的请求，而部署方法则确保相关的 verticle 能够初始化和运行。

@Componentpublic class RequestRouteDispatcher extends AbstractVerticle {

    @Override    public void start() throws Exception {        super.start();   vertx.eventBus().<String>consumer("CACHED") ❶        .handler(handleCachedRequest());                            vertx.eventBus().<String>consumer("PROXY")  ❷               .handler(handleProxyRequest());

        deployRedisManagerVerticle();  ❸        deployCollectManagerVerticle();❹}

❶ 为“CACHED”请求设置事件总线消费者：本行代码为地址为“CACHED”的消息设置事件总线消费者。每当消息到达时，它会调用handleCachedRequest方法。

❷ 为“PROXY”请求设置事件总线消费者：本行代码为地址为“PROXY”的消息设置事件总线消费者。每当消息到达时，它会调用handleProxyRequest方法。

❸ 部署 RedisManagerVerticle：本方法部署RedisManagerVerticle并记录部署结果。

❹ 部署 CollectManagerVerticle：本方法部署CollectManagerVerticle并记录部署结果。

Redis Manager

该类设置了一个事件总线消费者，通过检查 Redis 来处理缓存的请求，如果数据不存在的话，则可能会委托给其他的服务来进行处理。它还包含了一个刷新缓存的机制，确保数据定期更新。Redis 客户端配置为具备管理池和处理等待限制的选项，可优化在高负载情况下的性能。

import io.vertx.redis.client.Redis;

@Componentpublic class RedisClientManager extends AbstractVerticle {    private RedisAPI redisAPI;

    @Override    public void start() {      vertx.eventBus().<String>consumer("CACHED")❶                     .handler(handleRedisRequest());        vertx.setPeriodic(5000, id -> refershCache());❷ //refresh every 5 sec

      Redis client = Redis.createClient(vertx, new RedisOptions() ❸                .setMaxPoolSize(10)                .setMaxWaitingHandlers(50));      redisAPI = RedisAPI.api(client);      refershCache();    }private Handler<Message<String>> handleRedisRequest() { ❹    // Check if present in redis else delegate to proxy CollectAPI }

❶ 为“CACHED”请求设置事件总线消费者：本行代码为地址为“CACHED”的消息设置事件总线消费者。每当消息到达时，它会调用handleRedisRequest方法。

❷ 设置定期刷新缓存：本行代码设置了一个定期的任务，该任务会每 5000 毫秒（5 秒）调用一次refreshCache方法。

❸ 创建 Redis 客户端：本代码块使用指定的选项创建 Redis 客户端，并初始化redisAPI对象。

❹ 处理 Redis 请求：该方法检查请求的数据是否位于缓存中。如果存在的话，返回缓存的值。如果不存在，它将请求委托给“PROXY”地址并缓存结果。

Collect API Manager

这个类设置了一个事件总线消费者，用于处理对CollectAPI的代理请求。它会使用 Vert.x WebClient 向 API 发送 HTTP GET 请求，并添加必要的头信息，用于内容类型的确定和授权。在收到响应或遇到错误时，它会用相应的响应或错误消息回复原始消息。这有助于应用程序的不同部分以及外部CollectAPI服务之间的无缝集成和通信。

@Componentpublic class CollectAPIManager extends AbstractVerticle {

    public static final String COLLECT_API_URL = "api.collectapi.com";    private WebClient client;

    @Override    public void start() {    vertx.eventBus().<String>consumer("PROXY") ❶.handler(handleCollectAPIRequest());    client = WebClient.create(vertx, new WebClientOptions()); ❷    }

private Handler<Message<String>> handleCollectAPIRequest() { ❸       return msg -> {          client            .get(COLLECT_API_URL, msg.body())             .putHeader("content-type", "application/json")            .putHeader("authorization", "apikey your api key")            .send()            .onSuccess(response -> {   msg.reply(response.bodyAsString());            })            .onFailure(err -> {                msg.reply("Failed to connect CollectAPI" + err.getCause());            });        };    }

❶ 为 PROXY”请求设置事件总线消费者：本行代码为地址为“PROXY”的消息设置事件总线消费者。

❷ 创建 WebClient 实例：本行代码使用默认选项初始化一个新的WebClient实例。如果需要，可以自定义 WebClientOptions。

❸ 处理 CollectAPI 请求：本方法使用WebClient向CollectAPI发送 GET 请求。如果请求成功的话，它将以响应体作为回复。如果请求失败，则回复一条错误信息。

总结

本文深入探讨了 I/O 模型从阻塞式 I/O（BIO）到非阻塞式 I/O（NIO）和异步 I/O（AIO） 的演变过程及其对现代软件开发的影响，所解决的主要问题是传统阻塞式 I/O 模型在处理高负载和大量并发连接时的低效率和可扩展性限制，在云计算、大数据和物联网的时代，传统模型的不足变得越发明显。

我们所提出的解决方案是反应式架构，这种架构的目标是实现响应性、弹性、韧性和消息驱动。Vert.x 是在 JVM 上构建反应式应用程序的工具包，它提供了 Multi-Reactor 模式、事件总线和 Verticle 等功能，可高效处理高并发的问题，是一款功能强大的工具。基准测试结果表明，Vert.x 的性能优于其他工具，更加适合现代高性能应用程序。

本文提供了一个为 COVID-19 全球仪表盘构建 MultiReactor web 应用程序的详细案例研究。该应用使用 Vert.x 高效处理高吞吐、非阻塞的操作。它包括 Redis Manager 和 CollectAPI Manager 等组件，用于管理缓存数据和代理请求。MasterCard 和 Dream11 等公司的实际实施案例说明了 Vert.x 在构建可扩展高性能系统方面的实际应用和优势。这一全面分析证明了 Vert.x 在应对现代 I/O 和反应式应用程序开发挑战方面的有效性。

关于作者

Mohit Palriwal 是 Netflix 的高级软件工程师，是 Netflix Observability 团队的重要成员。他是 Netflix Atlas 项目团队成员，该项目是一个开源的多维时间序列数据库，旨在处理大规模场景下的需求。在加入 Netflix 之前，Mohit 是 Salesforce 的首席软件工程师，负责在 AWS 上构建 Observability Cloud。Mohit 的职业经历还包括了亚马逊云科技（AWS），在那里他花了四年多的时间开发并推出了无服务架构的 AWS Pinpoint。

查看英文原文：

https://www.infoq.com/articles/reactive-java-vertx-deep-dive/