Vert.x 基礎入門

Vert.x 是一個輕量級、高性能、模塊化的響應式編程技術，天生自帶異步、分布式屬性，簡約而不簡單，是目前實現微服務和 Serverless 架構的最熱門框架之一。

2011 年，在 VMware 工作的 Tim Fox 開始開發 Vert.x。2014 年 5 月，Vert.x 在 JAX 創新獎中榮獲“最具創新性的 Java 技術”獎。經過 8 年的發展，Vert.x 已經成長為一個功能全面的響應式編程框架，並且擁有了一個較為豐富的生態圈。通過使用 Vert.x 可以快速開發出高性能的 HTTP 服務和 WebSocket 服務，也可以作為底層網絡通信框架減少多線程編程的復雜性。

工程師對於性能的追求是無止境的，而響應式編程是提升系統性能的一大利器。追求極致的你是否願意跟我一起來一探究竟呢？

本文將帶你一起入門 Vert.x ，通過 Vert.x 是什么來理解這個工具的框架，然后再通過主要功能探究來學習主要功能，最后通過一個小實戰項目來鞏固所學的知識。文中提到的代碼都可以通過 https://github.com/roytrack/vertx-gitchat 來下載。下面我們開始吧~

Vert.x 是什么

在 Vert.x 的官網上 Vert.x 有這么一段話說明了它是什么。

Eclipse Vert.x is a tool-kit for building reactive applications on the JVM.

翻譯過來就是：Vert.x 是構建基於 JVM 的反應式應用的工具包。 首先，它是一個全家桶式的工具包，既可以用簡單的代碼來創建一個高性能的網絡服務器，也就是可以作為類似於 Apache 的 httpclient 來使用，提供 HTTP client 、DNS client 等一系列工具，同時提供各類好用的功能集合：

• 數據訪問：提供 MongoDB、JDBC、Redis 等客戶端；
• 集成其他框架：郵件客戶端、JCA 適配器、RabbitMQ 和 Kafka 客戶端、Consul 客戶端、MQTT 服務器；
• 事件總線橋接：Camel 和 TCP 的事件總線
• 登錄驗證集成：JDBC 驗證、JWT 驗證、Shiro 驗證、MongoDB 驗證、OAuth2 驗證
• 反應式編程：Vert.x Rx、反應式流處理、Vert.x Sync
• 微服務支持：服務發現、配置管理、熔斷器
• 開發運維：shell 和 docker 的支持、度量度量工具和包管理工具
• 異步測試支持：Vert.x Unit
• 集群支持：Hazelcast、Infinispan、Apache Ignite、Apache Zookeeper
• 雲支持：OpenShift Cartridge

其次，它可以用來構建反應式應用的，在其官網右上角有反應式宣言，聲明了反應式系統的特質：

1. 即時響應性
2. 回彈性
3. 彈性
4. 消息驅動

文中鏈接有對應宣言的翻譯解釋鏈接，限於篇幅，就不在此探討了。

框架定位

Vert.x 的定位非常靈活。

簡單的網絡工具、復雜的 web 應用程序、 HTTP / REST 微服務、海量事件處理、完整的后端消息總線應用程序都可以用 Vert.x 來做。

Vert.x 並不是一個嚴格意義上的框架或者容器，框架開發者不會告訴你一條“正確”的道路去寫一個應用，而是給你了很多強有力的工具，讓你去按照自己希望的方式去構建應用。

例如很多人在學習 Vert.x 的時候認為 Verticle 必不可少，其實可以只去用它的 HTTP client，而不用其他的，這也是可以的。

另外一個體現就是不強制要求只用 Vert.x 提供的功能，而是可以與其他類庫靈活的組合使用，例如 Spring 的 IOC 容器就可以靈活搭配起來。

框架功能簡介

Vert.x 生態系統是由一系列模塊組成的，一部分是官方維護，一部分是志願者維護。具體如下圖結構： 

Vert.x Core 作為最核心部分，提供了以下功能：

1. TCP 客戶端和服務器；
2. HTTP 客戶端和服務器，並且支持 WebSocket ，這部分我們會在下面常用兩個協議 ( HTTP & WebSocket ) 進一步展開講；
3. 事件總線，用來進行信息傳遞，實質是一種基於 TCP 連接的消息隊列，支持點對點和發布訂閱兩種機制；
4. 共享數據,通過本地的 map 結構和集群級別的分布式 map；
5. 定時器和延遲操作，通過 vertx.setPeriodic 實現周期定時執行， vertx.setTimer 來實現一次性的調用（延時操作）；
6. 裝載、卸載部署單元；
7. UDP 數據報文；
8. DNS 客戶端；
9. 訪問文件系統；
10. 高可用；
11. 集群化。

Vert.x Core 只包含一個非常輕量級的 jar 包，里面有以上的核心功能。

Vert.x Web 基於 Vert.x Core，提供了一系列更豐富的功能以便更容易地開發實際的 Web 應用。它繼承了 Vert.x 2.x 里的 Yoke 的特點，靈感來自於 Node.js 的框架 Express 和 Ruby 的框架 Sinatra 等等。Vert.x Web 的設計是強大的，非侵入式的，並且是完全可插拔的。Vert.x Web 非常適合編寫 RESTful HTTP 微服務。

Vert.x Stack 包含一系列提供鑒權、網絡、數據訪問等的官方擴展 jar 包。

而 Vert.x Awesome 則包含了官方和非官方提供的各種擴展，例如各種書籍、構建工具、雲支持、容器支持等。

下面放上一個一文全的地址: https://github.com/vert-x3/vertx-awesome ,可以體會一下 Vert.x 的生態規模。

黃金法則

Vert.x 中有一個黃金法則，就是不要阻塞 Event Loop 。

這個黃金法則設立的原因我們來一起探索一下。

大部分 Vert.x 的 API 都是事件驅動的,通過事件來調用起對應的 handler,這樣就可以解耦，進行異步操作。

例如下面的代碼示例，當有 HTTP 請求過來的時候，調用這個 handler 來返回響應信息：

server.requestHandler(request -> {
  // This handler will be called every time an HTTP request is received at the server request.response().end("hello world!"); }); 

相比於以前的處理，就是不會再阻塞處理，而是通過事件驅動，讓 Vert.x 自己主動調用。

這樣引入的優勢是，Vert.x 可以通過少量的線程，處理大量的並發事件。如果這里面並發事件同時出現了與線程數相同的阻塞操作，例如讀取文件，那么所有線程都被阻塞，整個程序就被掛起、拒絕服務了。

在大部分情況下，Vert.x 在一個線程里調用對應的 handlers，這個線程就叫做一個 Event Loop 。

圖片來自 Vert.x官網：https://vertx.io/docs/guide-for-java-devs/

Vert.x 保證自己的 API 是非阻塞的,並且不會阻塞 Event Loop，但是但是無法控制你自己的代碼是否是阻塞的,所以會對這部分進行監控，並通過日志警告你。我們來造一個阻塞操作看看：

public class BlockWarningDemo extends AbstractVerticle { @Override public void start() throws Exception { Thread.sleep(3000L); } public static void main(String[] args) { Vertx vertx = Vertx.vertx(); BlockWarningDemo blockWarningDemo = new BlockWarningDemo(); vertx.deployVerticle(blockWarningDemo); vertx.close(); } } 

運行上面的代碼就可以得到一個阻塞被警告的日志：

三月 03, 2019 11:08:06 上午 io.vertx.core.impl.BlockedThreadChecker
警告: Thread Thread[vert.x-eventloop-thread-0,5,main] has been blocked for 2847 ms, time limit is 2000 ms

默認 Event Loop 的線程執行時間是 2 秒，上面的代碼讓線程睡了三秒就直接被檢查出來了。與此類似的還有以下的阻塞操作：

1. 等待鎖；
2. 等待互斥鎖或者同步代碼塊；
3. 數據庫長時間查詢；
4. 進行消耗很多時間的大量計算；
5. 文件句柄的獲取與讀寫。

那么這些阻塞的操作我們怎么做呢？使用 vertx.executeBlocking 或者使用一類新的 Verticle ， 叫做 Worker Verticle 。

根本上就是讓這些阻塞操作在一組內部或者自定義的線程池上執行，而不要在處理事件的 Event Loop 操作。待處理完畢后，再通過 future 對應的 handler 來回調 Event Loop 進行后續處理。

通過這一部分的講解，我們知道了 Vert.x 是構建基於 JVM 的反應式應用的工具包，提供了一批可用於 TCP 、 UDP 、 HTTP 、 DNS 的工具集，具有較好的開源生態，並且對其黃金法則進行了探索與實戰。

主要功能探究

Vert.x 提供了很多功能，我們學習起來可能會有點選擇困難症，下面我挑了幾個比較核心的，能盡快形成生產力的功能來進行探索。

部署單元 ( Verticle )

在學習 Vert.x 的部署單元之前，我們先來理解一下 Actor 模型。

你可以將 Actor 當作是一群人，他們互相之間不會面對面地交流，而只是通過郵件的方式進行溝通。

Verticle 可以近似的看做 Actor 模型中的 actor （之所以說近似，因為各個 Verticle 還是可以通過共享 map 來通信。）。不同的 verticle 實體部署在一個 Vert.x 實例中，通過 Event Bus 來進行消息傳遞。

定義自己的 Verticle 需要擴展 AbstractVerticle ，邏輯可以寫在重載的不帶參數的 start 、 stop 方法中，表示在 Verticle 部署后，卸載后執行的邏輯。 如果重載帶參數 Future 的 start 、 stop 方法，則表示在部署或者卸載 Verticle 前要做一些邏輯，做完后將 future 設置為成功或者失敗，標志着 Verticle 相應動作成功或失敗。

Verticle 分為三類：

• 標准 Verticle ，在 Event Loop 上執行；
• 工作 Verticle ，在工作線程池運行。一個實例不會同時在多於一個線程上執行；
• 多線程工作 Verticle ，在工作線程池運行，是高階特性，一個實例不會同時在多於一個線程上執行，但是實例自身可以開啟多個線程。

Verticle 最大的特色就是它的線程隔離性。在啟動的時候， Verticle 就被分配給了創建和 start 方法調用的 Event Loop 了。當調用一個使用 core API 的 handler 的方法的時候， Vert.x 保證這些 handler 將在同一個 Event Loop 上執行的。

也就是說在 Verticle 實例的代碼保證是在同一個 Event Loop 執行。

上文中的 BlockWarningDemo 就是一個標准 Verticle 。如果要轉化為工作 Verticle ，在部署的時候進行以下配置：

DeploymentOptions options = new DeploymentOptions().setWorker(true); BlockWarningDemo blockWarningDemo = new BlockWarningDemo(); vertx.deployVerticle(blockWarningDemo, options); 

這樣這個 Verticle 就不在 Event Loop 上分配了，轉而分配到 Vert.x 的內部工作線程池上。

圖片來自 Vert.x官網：https://vertx.io/docs/guide-for-java-devs/

鏈式 API 與異步 ( Fluent API & Asynchronous )

鏈式操作和異步處理，是始終貫穿 Vert.x API 中的兩個通用模式。我們在編寫程序的時候更期望使用流暢的 API ，一路處理下來即可。例如使用 Builder 可以更方便的獲取新的對象：

package com.roytrack.fluent; import lombok.Builder; import lombok.ToString; @Builder @ToString public class LombokDemo { private final String userName; private final Integer age; public static void main(String[] args) { LombokDemo demo = LombokDemo.builder().userName("roy").age(30).build(); System.out.println(demo); } } 

亦或者是 Java 8 中引入的流處理 API ，更方便的進行流處理：

package com.roytrack.fluent; import java.util.ArrayList; import java.util.List; public class Java8Demo { public static void main(String[] args) { List<Integer> list = new ArrayList(); for (int i = 0; i < 100; i++) { list.add(i); } int result = list.stream().map(v -> v + 1).reduce(0, (v1, v2) -> v1 + v2); System.out.println(result); } } 

Vert.x 也大量的使用了類似的 API 設計，下面這個例子展示了它靈活的 Json 構造和處理：

package com.roytrack.fluent; import io.vertx.core.json.JsonObject; public class VertxDemo { public static void main(String[] args) { JsonObject jsonObject = new JsonObject().put("name", "roy").put("age", "30"); System.out.println(jsonObject.toString()); } } 

對於異步處理，大部分是通過設置 handler ，讓 Vert.x 自己根據響應去調用，舉例一個 request 的返回如下：

vertx.createHttpServer().requestHandler(req -> {
      req.response()
          .putHeader("content-type", "text/plain") .end("Hello from Vert.x!"); }); 

配合使用異步處理和鏈式 API 操作，我們既可以獲得代碼的簡潔性，也可以得到分布式下的高性能。

常用兩個協議 ( HTTP & WebSocket )

下面我們來寫一個最小的 HTTP 服務：

package com.roytrack.http; import io.vertx.core.Vertx; import io.vertx.core.http.HttpServer; public class MinimalHttpServer { public static void main(String[] args) { Vertx vertx = Vertx.vertx(); //創建 httpServer HttpServer server = vertx.createHttpServer().requestHandler(req -> { req.response() .putHeader("content-type", "text/plain") .end("Hello from Vert.x!"); }); //指定監聽端口 server.listen(8080, res -> { if (res.succeeded()) { System.out.println("Begin http server !"); } else { System.out.println("Http server occured error " + res.cause()); } }); } } 

訪問后獲得結果：  如果想同時構建 HTTP 和 WebSocket 服務，處理起來也很簡單，代碼如下：

package com.roytrack.http; import io.vertx.core.Vertx; public class HttpAndWsServer { public static void main(String[] args) { Vertx vertx = Vertx.vertx(); //創建 httpServer vertx.createHttpServer() //增加 websocket服務 .websocketHandler(ws -> { System.out.println("path is " + ws.path()); if (!ws.path().equals("/ws")) { ws.reject(); } ws.textMessageHandler(msg -> { ws.writeTextMessage(msg); }); }) //增加 http服務 .requestHandler(req -> { req.response() .putHeader("content-type", "text/plain") .end("Hello from Vert.x!"); }) //指定監聽端口 .listen(8080, res -> { if (res.succeeded()) { System.out.println("Begin http server !"); } else { System.out.println("Http server occured error " + res.cause()); } }); } } 

通過訪問 WebSocket 返回如下：  同時，訪問 HTTP 也不受影響，因為 WebSocket 協議的升級包也是 HTTP 協議。

另外如果使用 Verticle 多實例部署，也可以共用一個端口，這樣一個 Verticle 停用或者卸載了，也不影響其他 Verticle 的服務，體現了反應式宣言中的回彈性。

多語言編程 ( Polyglot )

Vert.x 不止支持 Java，還官方支持 Ceylon 、 Grovvy 、 JavaScript 、Ruby 、 Scala 、 Kotlin ，貢獻者支持 Python 、 TypeScript 。

最快體驗多語言特性，莫過於安裝一下 Vert.x 的命令行工具，下面是兩種安裝方式。

通過 Node 安裝 vertx ，如果是 windows 系統，可以進入到對應 module 中執行 vertx.bat：

npm install vertx3-full

或者MacOS 通過 Homebrew 安裝 vertx ：

brew install vert.x

安裝后就可以執行 vertx 命令了。

編寫一個 js 腳本 my-verticle.js 如下：

var server = vertx.createHttpServer(); server.requestHandler(function (request) { request.response().end("Hello world"); }); server.listen(8080); 

通過執行以下命令，就可以啟動一個 httpServer 返回 Hello world 了。

vertx run my-verticle.js

項目實戰 ：構建一個資源管理器

經過以上的學習，我們對 Vert.x 的主要功能有一些了解了，下面我們來實戰一個項目。

需求：調用 MXBean 來獲取 cpu 和內存，然后通過 eventbus 發送。另外一個接收 eventbus ，發送到頁面進行展現。 最終效果如下圖：  最終完成的項目地址為：https://github.com/roytrack/vertx-gitchat

構建 Agent

com.roytrack.dashboard.Agent 核心代碼如下：

  @Override public void start() { OperatingSystemMXBean systemMBean = ManagementFactory.getPlatformMXBean(OperatingSystemMXBean.class); String pid = UUID.randomUUID().toString(); vertx.setPeriodic(1000, id -> { JsonObject metrics = new JsonObject(); metrics.put("CPU", systemMBean.getProcessCpuLoad()); metrics.put("Mem", systemMBean.getTotalPhysicalMemorySize() - systemMBean.getFreePhysicalMemorySize()); vertx.eventBus().publish("metrics", new JsonObject().put(pid, metrics)); }); } 

獲取系統的 MBean ，設置每秒執行一次，將相關參數放入到 JsonObject ，然后通過 EventBus 發送到 metrics 這個地址。

構建 Server

com.roytrack.dashboard.Server 核心代碼如下：

  @Override public void start() { Router router = Router.router(vertx); // The web server handler router.route().handler(StaticHandler.create().setCachingEnabled(false)); router.get("/dashboard").handler(ctx -> { ctx.response() .putHeader("Content-Type", "application/json") .end(dashboard.encode()); }); vertx.eventBus().<JsonObject>consumer("metrics").handler(msg -> { JsonObject metrics = msg.body(); dashboard.mergeIn(metrics); }); vertx.createHttpServer() .requestHandler(router::accept) .listen(8080); } 

StaticHandler 會去找 webroot 的靜態資源，直接對外提供服務。

/dashboard 路徑每次返回最新的數據。

消費地址為 metrics 的 EventBus ，來更新 Json 對象 dashboard 的內容。

最后創建 httpServer，監聽 8080 端口即可。

項目代碼下載下來的同學，會發現給出的項目代碼還使用了 hazelcast 作為集群發現的方式，可以作為一個進階增強進行自學，有不懂的地方我們在讀者圈繼續交流。

學習資料

1. 我們可以通過 Vert.x 的官方文檔系統學習它，地址為：https://vertx.io/docs/

2. 同時github上有人翻譯了中文版：https://vertxchina.github.io/vertx-translation-chinese/

3. 官網同時有一本電子書《使用Java構建響應式微服務》，大家可以學習參考：https://developers.redhat.com/promotions/building-reactive-microservices-in-java/

4. Vert.x生態的各種信息和例子，都可以從這個項目里找到：https://github.com/vert-x3/vertx-awesome