浅入Vert.x | Geepair’s Blog

type

Post

status

Published

date

Nov 27, 2023

slug

summary

1.什么是Vert.x？

1.1简介

Eclipse Vert.x

Vert.x | Reactive applications on the JVM

https://vertx.io/

Vert.x 一个基于JVM、轻量级、高性能的应用平台。

1.2性能表现

TechEmpower框架基准项目：执行基本任务（如JSON序列化，数据库访问和服务器端模板组合）的许多Web应用程序框架的性能比较。

TechEmpower Framework Benchmarks

Performance comparison of web application frameworks using community-contributed test implementations.

https://www.techempower.com/benchmarks/

单条SQL查询排名第6

复合SQL查询排名第5

全表SQL查询排名第5

1.3为什么高性能

1.3.1与Tomcat的对比

Tomcat使用了同步阻塞式的I/O模型来处理请求，在这种模型下，当服务器接收到一个请求后，这个接收请求线程会去处理该请求，执行相关任务，如果它需要访问服务端I/O资源，那么调用者只有等待执行结果。

Vert.x本身是事件驱动（Reactor）、非阻塞纯异步IO模型，这意味着可以使用很少的线程资源处理大量并发请求。

Vertx实现了Multi-Reactor模式，区别于单线程的Reactor模式(Node.js实现了这种模式)。单一线程的问题在于它在任意时刻只能运行在一个核上，如果你希望单线程（Reactor）应用（如你的 Node.js 应用）扩展到多核服务器上，则需要启动并且管理多个不同的进程。

Vert.x的工作方式有所不同，每个Vertx实例维护的是多个Event Loop线程。默认情况下，会根据机器上可用的核数量来设置Event Loop的数量，也可自行设置。

Vert.x高性能的原因是支持让我们以异步非阻塞的方式来编写业务逻辑，且可以充分利用服务器的多核。

1.3.2同步阻塞式编程

大多数应用程序和服务开发框架都基于多线程。每个连接对应一个线程，开发者可以使用传统的命令式代码。

存在的问题

当使用传统的阻塞式API做以下操作时，调用线程可能会被阻塞：

从Socket中读取数据

写数据到磁盘

发送消息给接收者并等待回复

……

这意味着如果使用阻塞式API处理大量并发，需要大量线程来防止应用程序停止运转。而这些线程使用的内存（例如它们的栈）和线程上下文切换带来昂贵的开销。

缺点

工作负载过高，多线程处理这些请求会产生太多的上下文切换动作

线程池中处于IO 阻塞的线程不能被复用（处理新的Http请求）

需要解决多线程进行内存访问所存在的线程安全问题

阻塞式的方式对于现代应用程序所需要的并发级别来说是难于扩展的

1.3.3Vertx异步非阻塞式编程

使用 Asynchronous I/O，可以用更小的线程处理更多的连接。当一个任务发生了I/O操作时， Asynchronous I/O不会阻塞线程，而是执行其他待处理的任务，待到I/O结果准备好后再继续执行该任务。

Vert.x使用事件循环（Event Loop）多路复用并发处理工作负载。

在Event Loop上运行的代码不应执行阻塞I/O及长时间的处理逻辑。如果你只有单个 Event Loop，而且你希望每秒处理10000个 HTTP 请求，很明显的是每一个请求处理时间不可以超过0.1毫秒，所以你不能阻塞任何过多（大于0.1毫秒）的时间。

Vert.x可以只使用少量的线程来处理大量的并发（不会阻塞）。一个单独的Event Loop可以非常迅速地处理数千个HTTP请求。每个Vertx实例维护的是多个Event Loop线程。默认情况下，会根据机器上可用的核数量来设置Event Loop的数量。

缺点

采用同步命令式编程，符合人类的思维方式，而绝大多数异步编程，与我们人类思维上存在差异。异步编程的代码在阅读和理解上相比更困难。

可与Kotlin协程结合使用（await），用同步的方式编写异步代码

面向对象 or 函数式编程

1.4响应式编程为什么没有成为主流

思维差异+可维护性差（面向对象及其过程）

生态较差（对比Java语言的响应式类库RxJava）

1.5Vert.x在响应式编程的改善

回调地狱

配合Kotlin协程改善了回调地狱的现象，在JS的世界中，早已出现 Promise与await来解决这个问题。将非阻塞回调转成同步风格但实质还是非阻塞。

较为完整的生态

在Web,数据库，单元测试，权限，微服务支持，消息事件机制,集群等有完整的解决方案。

2.示例

开始使用要求

JDK 1.8 或更高版本

文本编辑器或 IDE

Maven 3 或更高版本

创建一个Http Server


# vertex-web包 4.5.0
<dependency>
      <groupId>io.vertx</groupId>
      <artifactId>vertx-core</artifactId>
      <version>${vertx.version}</version>
</dependency>


public class MainVerticle extends AbstractVerticle {

    public static void main(String[] args) {
        Launcher.executeCommand("run", MainVerticle.class.getName());
    }

    @Override
    public void start() throws Exception {
        vertx.createHttpServer().requestHandler(req -> {
            req.response().end("hello world");
        }).listen(8080, res -> {
            if (res.failed()) {
                System.out.println("start fail...");
                    res.cause().printStackTrace();
            } else {
                System.out.println("start success...");
            }
        });
    }
}

创建一个Web Client


public class ClientVerticle extends AbstractVerticle {

    public static void main(String[] args) {
        Launcher.executeCommand("run", ClientVerticle.class.getName());
    }

    @Override
    public void start() throws Exception {

        WebClient client = WebClient.create(vertx);

        client.get(8080, "localhost", "/").send(ar -> {
            if (ar.succeeded()) {
                HttpResponse<Buffer> response = ar.result();
                System.out.println("Got HTTP response with status " + response.statusCode() + " with data " + response.body().toString("UTF-8"));
            } else {
                ar.cause().printStackTrace();
            }
        });
    }
}

点对点消息


public class MessageVerticle1 extends AbstractVerticle {

    public static void main(String[] args) {
        VertxOptions options = new VertxOptions();
        Vertx.clusteredVertx(options, res -> {
            if (res.succeeded()) {
                Vertx vertx = res.result();
                vertx.deployVerticle("live.javaer.example.message.MessageVerticle1");
            }
        });
    }


    @Override
    public void start() throws Exception {
        Random random = new Random();
        vertx.setPeriodic(1000, t -> {
            vertx.eventBus().publish("message", random.nextInt());
        });
    }
}


public class MessageVerticle2 extends AbstractVerticle {

    public static void main(String[] args) {
        VertxOptions options = new VertxOptions();
        Vertx.clusteredVertx(options, res -> {
            if (res.succeeded()) {
                Vertx vertx = res.result();
                vertx.deployVerticle("live.javaer.example.message.MessageVerticle2");
            }
        });
    }


    @Override
    public void start() throws Exception {
        vertx.eventBus().consumer("message", msg -> {
            System.out.println("consumer message: " + msg.body());
        });
        System.out.println("consumer ready!");
    }
}