Vert.x，Databases

Vert.x数据库(访问)简介

Vert.x提供了数据库相关模块(扩展包)，方便我们通过异步方式访问数据库。数据库扩展主要包括：响应式数据库客户端(vertx-xxx-client)和通用JDBC客户端(vertx-jdbc-client)。

响应式数据库客户端是事件驱动和非阻塞的，可以仅仅使用一条线程来处理大量的数据库连接。专注于可伸缩性和低开销。这些客户端还内置连接池，提供了更简单易用的API。目前(v4.5.10)，Vert.x提供主流数据库的响应式客户端，包含: PostgreSQL，MySQL，DB2，Oracle，MSSQL，MongoDB，Redis，Apache Cassandra。

如果不想使用Vert.x提供的响应式数据库客户端(比如某些特性不支持)，或者Vert.x还没提供你所需要访问的数据库的响应式客户端，如: H2，那么可以通过通用JDBC客户端加JDBC驱动方式异步访问对应数据库。

Vert.x通过Sql Client API(vertx-sql-client)使用数据库客户端客户端，类图大致如下:

SQLOperations  <---   SQLClient <--  JDBCClient  <-- Pool <-- JDBCPool <--  JDBCPoolImpl<--+\- SQLConnection

使用响应式客户端

以MySQL为例，我们首先需要引入vertx-mysql-client依赖:

<dependency><groupId>io.vertx</groupId><artifactId>vertx-mysql-client</artifactId><version>4.5.10</version>
</dependency>

MySQL客户端的案例:

import java.util.logging.Level;
import java.util.logging.Logger;import io.vertx.core.Future;
import io.vertx.mysqlclient.MySQLBuilder;
import io.vertx.mysqlclient.MySQLConnectOptions;
import io.vertx.sqlclient.PoolOptions;
import io.vertx.sqlclient.Row;
import io.vertx.sqlclient.RowSet;
import io.vertx.sqlclient.SqlClient;
import io.vertx.sqlclient.Tuple;public class Db1 {private static final Logger LOGGER = Logger.getLogger(Db1.class.getName());public static void main(String[] args) {MySQLConnectOptions connectOptions = new MySQLConnectOptions().setHost("127.0.0.1").setPort(3306).setUser("root").setPassword("Passw0rd").setDatabase("information_schema").addProperty("autoReconnect", "true") //添加连接参数.addProperty("useSSL","false").addProperty("rewriteBatchedStatements", "true");PoolOptions poolOptions = new PoolOptions().setMaxSize(5);SqlClient client = MySQLBuilder.client().with(poolOptions).connectingTo(connectOptions).build();String sqlText = "select table_name, TABLE_ROWS from tables where table_type=? and table_schema=? limit 2";Future<RowSet<Row>> future = client.preparedQuery(sqlText).execute(Tuple.of("BASE TABLE", "performance_schema"));future.onComplete(ar -> {if (ar.succeeded()) {RowSet<Row> rows = ar.result();for (Row row : rows) {// 注意: 字段名区分大小写// 否则抛异常: java.util.NoSuchElementException: Column xxx does not existLOGGER.info(row.getString("TABLE_NAME") + ", " + row.getLong("TABLE_ROWS") + " rows.");}} else {LOGGER.log(Level.SEVERE, "", ar.cause());}client.close(); // 关闭客户端});}
}

使用JDBC客户端

要使用JDBC客户端，需要引入"vertx-jdbc-client"和对应的JDBC驱动，如果使用连接池，还需要引入"agroal-pool"依赖：

<dependency><groupId>io.vertx</groupId><artifactId>vertx-jdbc-client</artifactId><version>4.5.10</version>
</dependency>
<dependency><groupId>io.agroal</groupId><artifactId>agroal-pool</artifactId><version>2.5</version>
</dependency>
<dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId><version>8.4.0</version>
</dependency>

JdbcClient访问MySQL数据库

访问MySQL数据库案例：

import java.util.logging.Level;
import java.util.logging.Logger;import io.vertx.core.Vertx;
import io.vertx.jdbcclient.JDBCConnectOptions;
import io.vertx.jdbcclient.JDBCPool;
import io.vertx.sqlclient.PoolOptions;
import io.vertx.sqlclient.Row;
import io.vertx.sqlclient.RowSet;
import io.vertx.sqlclient.Tuple;public class Db2 {private static final Logger LOGGER = Logger.getLogger(Db2.class.getName());public static void main(String[] args) {JDBCConnectOptions connectOptions = new JDBCConnectOptions().setJdbcUrl("jdbc:mysql://localhost:3306/information_schema?useSSL=false&autoReconnect=true").setUser("root").setPassword("Passw0rd");PoolOptions poolOptions = new PoolOptions().setMaxSize(5).setName("mysql-conn-pool");Vertx vertx = Vertx.vertx();String sqlText = "select table_name, TABLE_ROWS from tables where table_type=? and table_schema=? limit 2";JDBCPool client = JDBCPool.pool(vertx, connectOptions, poolOptions);client.preparedQuery(sqlText).execute(Tuple.of("BASE TABLE", "performance_schema")).onComplete(ar -> {if (ar.succeeded()) {RowSet<Row> rows = ar.result();for (Row row : rows) {// 注意: 字段名区分大小写// 否则抛异常: java.util.NoSuchElementException: Column xxx does not existLOGGER.info(row.getString("TABLE_NAME") + ", " + row.getLong("TABLE_ROWS") + " rows.");}} else {LOGGER.log(Level.SEVERE, "", ar.cause());}client.close(); // 关闭客户端});}
}

Retrieving the generated keys

获取自动生成的键值。该功能依赖JDBC驱动的功能。

// 测试表结构: create table tpcc.test1(id int not null auto_increment, name varchar(32), primary key (id));
JDBCPool client = ...
// SqlClient = ... MySQL异步客户端不支持以下用法, lastInsertId.getLong(0)会返回空值。
String sqlText = "insert tpcc.test1(name) values('hello mysql client!')";
client.preparedQuery(sqlText).execute().onSuccess(rows -> {Row lastInsertId = rows.property(JDBCPool.GENERATED_KEYS);long newId = lastInsertId.getLong(0);LOGGER.info("generated keys: " + newId);client.close();
}).onFailure(e->e.printStackTrace());

使用存储过程和函数

/*
delimiter $$
create procedure tpcc.new_test1(OUT id INT, IN  name VARCHAR(32))
begininsert tpcc.test1(name) values(name);select LAST_INSERT_ID() into id;
end $$
delimiter ;
*/
String sqlText = "{call tpcc.new_test1(?, ?)}";
client.preparedQuery(sqlText).execute(Tuple.of(SqlOutParam.OUT(JDBCType.INTEGER), "John")).onFailure(e -> e.printStackTrace()).onSuccess(rows -> {if (rows.property(JDBCPool.OUTPUT)) {for (Row row : rows) {LOGGER.info("generated keys: " + row.getLong(0));}}client.close();});

SQL客户端模版

Vert.x提供一小的工具库SQL Client Templates来便利我们对数据库的访问，要使用SQL客户端模版需要引入vertx-sql-client-templates依赖:

<dependency><groupId>io.vertx</groupId><artifactId>vertx-sql-client-templates</artifactId><version>4.5.10</version>
</dependency>

使用命名参数

SqlClient client = ...
Map<String, Object> parameters = new HashMap<>();
parameters.put("type", "BASE TABLE");
parameters.put("schema", "performance_schema");String sqlText = "select table_name, TABLE_ROWS from tables where table_type=#{type} and table_schema=#{schema} limit 2";
SqlTemplate.forQuery(client, sqlText).execute(parameters).onSuccess(rows -> {for (Row row : rows) {LOGGER.info(row.getString("TABLE_NAME") + ", " + row.getLong("TABLE_ROWS") + " rows.");}client.close();
});

行映射

可以提供一个自定义的RowMapper，将Row映射到一个实体类。

//   Row -> 用户类PsTable
class PsTable {String tableName;int rows;public String getTableName() {return tableName;}public void setTableName(String name) {tableName = name;}public void setRows(int rows) {this.rows = rows;}public int getRows() {return rows;}@Overridepublic String toString() {return tableName + ", " + rows + " rows.";}
}
// ...
SqlTemplate.forQuery(client, sqlText).mapTo(row -> {PsTable table = new PsTable();table.setTableName(row.getString("TABLE_NAME"));table.setRows(row.getInteger("TABLE_ROWS"));return table;
}).execute(parameters).onSuccess(tables -> {for (PsTable table : tables) {LOGGER.info(table.toString());}client.close();
});