實時必須

實時處理領域，當需要使用外部儲存資料染色的時候，需要慎重對待，不能讓與外部系統之間的互動延遲對流的整個進度取決定性的影響。

同步的與資料庫互動需要等待一個請求從發起到結束才能發起下次請求，等待過程是非常浪費函式時間的。

與資料庫非同步互動，意味著單個函式例項可以併發處理很多請求，同時併發接收響應。那麼，等待時間由於傳送其它請求和接收其它響應，被重複使用而節省了。至少，等待時間在多個請求上被攤銷。這就使得很多使用案例具有更高的吞吐量。

浪尖在這裡推薦這個jdbc非同步訪問客戶端的主要目的是昨天發了一篇flink非同步IO的文章

這個非同步IO是實現高效能維表的一種方案，而今天要說的這個jdbc客戶端也是其中比較重要的一個環節。

Vert.x JDBC client客戶端的github地址

https://vertx.io/docs/vertx-jdbc-client/java/

使用該客戶端要加入的maven依賴配置到pom.xml檔案中

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-jdbc-client</artifactId>
  <version>3.5.2</version>
</dependency>

<dependency>
  <groupId>

大多數情況下可能是希望在多個客戶端例項中共享資料來源。

例如，通過建立多個例項來擴充套件應用程式，並且希望每個例項共享相同的資料來源，避免建立多個連結池。

Client

實現起來很簡單，只需呼叫：

SQLClient client =JDBCClient.createShared(vertx, config);

第一次呼叫該方法確實會使用指定的配置建立一個數據源。

後續的呼叫會直接返回一個客戶端例項，配置不會再次使用。

可以為建立的資料來源指定一個名稱，方式如下：

SQLClient client =JDBCClient.createShared(vertx, config, "MyDataSource");

如果使用相同的資料來源名稱和相同的Vert.x例項建立不同的客戶端，實際上客戶端會共享相同的資料來源。

這種建立方式是在你希望有不同的客戶端組分別去跟不同的資料庫互動的時候使用。

當然了，有時候也可能是希望每個客戶端例項獨享一個數據源，這種客戶端的建立方式也很簡單。

SQLClient client =JDBCClient.createNonShared(vertx, config);

這種方式等價於每次呼叫客戶端給一個唯一的資料來源名稱。

如果已經存在了一個數據源，那麼也可以直接指定該資料來源

SQLClient client = JDBCClient.create(vertx,dataSource);

在使用完了資料庫客戶端之後應該釋放掉資源。共享資料來源的客戶端會存在一個引用計數的概念。相同資料來源的最後一個引用關閉的時候，資料來源才會被釋放掉。

Connection

使用客戶端物件的getConnection方法可以獲取一個數據庫連結，具體使用方式如下：

client.getConnection(res -> {

 if(res.succeeded()) {



  SQLConnection connection = res.result();



  connection.query("SELECT * FROM some_table", res2 -> {

    if (res2.succeeded()) {



      ResultSet rs = res2.result();

      // Do something with results

    }

  });

 }else {

  // Failed to get connection - deal with it

 }

});

SQLConnection是一個通用的介面，不僅僅適用於Vert.x客戶端。

介面的介紹如下：

https://vertx.io/docs/vertx-sql-common/java/

配置

主要是介紹一下常用的配置。

provider_class ：該類主要是用來管理資料庫連線的。預設是

io.vertx.ext.jdbc.spi.impl.C3P0DataSourceProvider。可以覆蓋該引數配置，指定自己的實現。已實現的方式多種，可以參考

https://github.com/vert-x3/vertx-jdbc-client/tree/master/src/test/java/io/vertx/ext/jdbc/spi/impl

row_stream_fetch_size : SQLRowStream的內部cache的大小，預設是128.

假設使用了C3P0的實現，也即是預設，配置引數如下：

url : 資料庫連線用的URL。

driver_class : jdbc driver的類名稱。

user:資料庫的使用者名稱。

password:資料庫密碼。

max_pool_size:連線池中最大連線數量，預設是15.

initial_pool_size:連線池中最大初始連線數量，預設是3.

min_pool_size:連線池中最小的連線數量。

max_statements:快取的prepared statementszui 大數量，預設是0.

max_statements_per_connection:每個連線快取的prepared statements最大的數量，預設是0。

max_idle_time: 一個連線的空閒時間超過該值，就會被關閉，預設是0，禁止超時關閉機制.

完整的案例

package bigdata.spark.DaoUtil;

import com.google.common.collect.Lists;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.jdbc.JDBCClient;
import io.vertx.ext.sql.ResultSet;
import io.vertx.ext.sql.SQLClient;
import io.vertx.ext.sql.SQLConnection;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.types.Row;

import java.util.Collections;
import java.util.List;

public class VertxMysql {
   private transient SQLClient mySQLClient;

   public void open() throws Exception {
       JsonObject mySQLClientConfig = new JsonObject();
       mySQLClientConfig.put("url", "jdbc:mysql://localhost:3306/")
               .put("driver_class", "com.mysql.jdbc.Driver")
               .put("max_pool_size", 20)
               .put("user", "root")
               .put("password", "password");

       VertxOptions vo = new VertxOptions();
       vo.setEventLoopPoolSize(10);
       vo.setWorkerPoolSize(20);
       Vertx vertx = Vertx.vertx(vo);
       mySQLClient = JDBCClient.createNonShared(vertx, mySQLClientConfig);
   }

   public void close() throws Exception {
       mySQLClient.close();
   }

   public void asyncInvoke() throws Exception {

//        JsonArray inputParams = new JsonArray();

       mySQLClient.getConnection(conn -> {
           if (conn.failed()) {
               //Treatment failures
               return;
           }

           final SQLConnection connection = conn.result();
           connection.query("SELECT * FROM some_table", res2 -> {
               if (res2.succeeded()) {

                   ResultSet rs = res2.result();
                   List<JsonObject> rows = rs.getRows();
                   for(JsonObject json : rows){
                       String id = json.getString("id");
                       String name = json.getString("name");

                       System.out.println("name : "+name + "ID : "+ id);
                   }
                   // Do something with results
               }
           });

       });
   }
}

github地址:

https://github.com/vert-x3/vertx-jdbc-client

推薦閱讀：