响应式 MSSQL 客户端

The client is reactive and non-blocking, allowing to handle many database connections with a single thread.

< groupId > io.vertx </ groupId > < artifactId > vertx-mssql-client </ artifactId > < version > 4.4.0 </ version > </ dependency >

MSSQLConnectOptions connectOptions = new MSSQLConnectOptions()
  .setPort(1433)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");
// 连接池参数
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);
// 创建客户端池
MSSQLPool client = MSSQLPool.pool(connectOptions, poolOptions);
// 一个简单的查询
client
  .query("SELECT * FROM users WHERE id='julien'")
  .execute(ar -> {
  if (ar.succeeded()) {
    RowSet result = ar.result();
    System.out.println("Got " + result.size() + " rows ");
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  // 现在关闭池
  client.close();
MSSQLConnectOptions connectOptions = new MSSQLConnectOptions()
  .setPort(1433)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");
// 池参数
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);
// 创建池化客户端
MSSQLPool client = MSSQLPool.pool(connectOptions, poolOptions);
池化客户端使用连接池，任何操作都会从池中借用连接，
随后执行操作，并最终执行完之后将其释放到池中。
如果您使用 Vert.x 运行，您可以将它传递给您的 Vertx 实例：
MSSQLConnectOptions connectOptions = new MSSQLConnectOptions()
  .setPort(1433)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");
// 池选项
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);
// 创建池化客户端
MSSQLPool client = MSSQLPool.pool(vertx, connectOptions, poolOptions);
当您不再需要它时，您需要释放池：
pool.close();
当您需要在同一个连接上执行多个操作时，需要使用客户端
您可以轻松地从池中获取一个连接：
MSSQLConnectOptions connectOptions = new MSSQLConnectOptions()
  .setPort(1433)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");
// 池选项
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);
// 创建池化客户端
MSSQLPool client = MSSQLPool.pool(vertx, connectOptions, poolOptions);
// 从池中获取连接
client.getConnection().compose(conn -> {
  System.out.println("Got a connection from the pool");
  // 所有操作都在同一个连接上执行
  return conn
    .query("SELECT * FROM users WHERE id='julien'")
    .execute()
    .compose(res -> conn
      .query("SELECT * FROM users WHERE id='emad'")
      .execute())
    .onComplete(ar -> {
      // 释放连接并将其归还给池
      conn.close();
}).onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Done");
  } else {
    System.out.println("Something went wrong " + ar.cause().getMessage());
完成连接后，您必须关闭它以将其释放到池中，以便可以重复使用。
MSSQLConnectOptions connectOptions = new MSSQLConnectOptions()
  .setPort(1433)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");
// 池参数
PoolOptions poolOptions = new PoolOptions().setMaxSize(5);
// 根据数据对象创建池
MSSQLPool pool = MSSQLPool.pool(vertx, connectOptions, poolOptions);
pool.getConnection(ar -> {
  // 处理您的连接
String connectionUri = "sqlserver://dbuser:[email protected]:1433/mydb";
// 从连接 URI 创建池
MSSQLPool pool = MSSQLPool.pool(connectionUri);
// 从连接 URI 创建连接
MSSQLConnection.connect(vertx, connectionUri, res -> {
  // 处理您的连接
连接 URI 格式由客户端以惯用方式定义：
sqlserver://[user[:[password]]@]host[:port][/database][?<key1>=<value1>[&<key2>=<value2>]]
当前，客户端在连接 uri 中支持以下参数关键字（key不区分大小写）：




    

  if (ar.succeeded()) {
    RowSet<Row> result = ar.result();
    System.out.println("Got " + result.size() + " rows ");
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
执行预查询也是一样的操作。
SQL字符通过位置引用实际的参数，并使用数据库的语法 `@p1`, `@p2`, etc…
client
  .preparedQuery("SELECT * FROM users WHERE id=@p1")
  .execute(Tuple.of("julien"), ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    System.out.println("Got " + rows.size() + " rows ");
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
查询相关的方法为 SELECT 类型的操作提供了异步的 RowSet 实例
client
  .preparedQuery("SELECT first_name, last_name FROM users")
  .execute(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("User " + row.getString(0) + " " + row.getString(1));
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
或者 UPDATE/INSERT 类型的查询：
client
  .preparedQuery("INSERT INTO users (first_name, last_name) VALUES (@p1, @p2)")
  .execute(Tuple.of("Julien", "Viet"), ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    System.out.println(rows.rowCount());
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
Row对象（Row）可以让您通过索引位置获取相应的数据
System.out.println("User " + row.getString(0) + " " + row.getString(1));
String firstName = row.getString("first_name");
Boolean male = row.getBoolean("male");
Integer age = row.getInteger("age");
您可以使用缓存过的预处理语句去执行一次性的预查询：
connectOptions.setCachePreparedStatements(true);
client
  .preparedQuery("SELECT * FROM users WHERE id = @p1")
  .execute(Tuple.of("julien"), ar -> {
    if (ar.succeeded()) {
      RowSet<Row> rows = ar.result();
      System.out.println("Got " + rows.size() + " rows ");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
您也可以创建 PreparedStatement 并自主地管理它的生命周期。
sqlConnection
  .prepare("SELECT * FROM users WHERE id = @p1", ar -> {
    if (ar.succeeded()) {
      PreparedStatement preparedStatement = ar.result();
      preparedStatement.query()
        .execute(Tuple.of("julien"), ar2 -> {
          if (ar2.succeeded()) {
            RowSet<Row> rows = ar2.result();
            System.out.println("Got " + rows.size() + " rows ");
            preparedStatement.close();
          } else {
            System.out.println("Failure: " + ar2.cause().getMessage());
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
List<Tuple> batch = new ArrayList<>();
batch.add(Tuple.of("julien", "Julien Viet"));
batch.add(Tuple.of("emad", "Emad Alblueshi"));
// Execute the prepared batch
client
  .preparedQuery("INSERT INTO USERS (id, name) VALUES (@p1, @p2)")
  .executeBatch(batch, res -> {
  if (res.succeeded()) {
    // Process rows
    RowSet<Row> rows = res.result();
  } else {
    System.out.println("Batch failed " + res.cause());
client
  .preparedQuery("INSERT INTO movies (title) OUTPUT INSERTED.id VALUES (@p1)")
  .execute(Tuple.of("The Man Who Knew Too Much"), res -> {
    if (res.succeeded()) {
      Row row = res.result().iterator().next();
      System.out.println(row.getLong("id"));
  .compose(connection ->
    connection
      .preparedQuery("INSERT INTO Users (first_name,last_name) VALUES (@p1, @p2)")
      .executeBatch(Arrays.asList(
        Tuple.of("Julien", "Viet"),
        Tuple.of("Emad", "Alblueshi")
      .compose(res -> connection
        // Do something with rows
        .query("SELECT COUNT(*) FROM Users")
        .execute()
        .map(rows -> rows.iterator().next().getInteger(0)))
      // Return the connection to the pool
      .eventually(v -> connection.close())
  ).onSuccess(count -> {
  System.out.println("Insert users, now the number of users is " + count);
也可以通过连接对象创建预查询:
connection
  .prepare("SELECT * FROM users WHERE first_name LIKE @p1")
  .compose(pq ->
    pq.query()
      .execute(Tuple.of("Julien"))
      .eventually(v -> pq.close())
  ).onSuccess(rows -> {
  // All rows
pool.withConnection(connection ->
  connection
    .preparedQuery("INSERT INTO Users (first_name,last_name) VALUES (@p1, @p2)")
    .executeBatch(Arrays.asList(
      Tuple.of("Julien", "Viet"),
      Tuple.of("Emad", "Alblueshi")
    .compose(res -> connection
      // Do something with rows
      .query("SELECT COUNT(*) FROM Users")
      .execute()
      .map(rows -> rows.iterator().next().getInteger(0)))
).onSuccess(count -> {
  System.out.println("Insert users, now the number of users is " + count);
      .compose(tx -> conn
        // Various statements
        .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
        .execute()
        .compose(res2 -> conn
          .query("INSERT INTO Users (first_name,last_name) VALUES ('Andy','Guibert')")
          .execute())
        // Commit the transaction
        .compose(res3 -> tx.commit()))
      // Return the connection to the pool
      .eventually(v -> conn.close())
      .onSuccess(v -> System.out.println("Transaction succeeded"))
      .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
当数据库服务端返回当前事务已失败（比如常见的 current transaction is aborted, commands ignored until end of transaction block）
，事务已回滚和 completion 方法的返回值future返回了
TransactionRollbackException 异常时：
tx.completion()
  .onFailure(err -> {
    System.out.println("Transaction failed => rolled back");
pool.withTransaction(client -> client
  .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
  .execute()
  .flatMap(res -> client
    .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
    .execute()
    // Map to a message result
    .map("Users inserted")))
  .onSuccess(v -> System.out.println("Transaction succeeded"))
  .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
connection.prepare("SELECT * FROM users WHERE age > @p1", ar1 -> {
  if (ar1.succeeded()) {
    PreparedStatement pq = ar1.result();
    // Create a cursor
    Cursor cursor = pq.cursor(Tuple.of(18));
    // Read 50 rows
    cursor.read(50, ar2 -> {
      if (ar2.succeeded()) {
        RowSet<Row> rows = ar2.result();
        // Check for more ?
        if (cursor.hasMore()) {
          // Repeat the process...
        } else {
          // No more rows - close the cursor
          cursor.close();
游标释放时需要同时执行关闭操作：
cursor.read(50, ar2 -> {
  if (ar2.succeeded()) {
    // Close the cursor
    cursor.close();
stream API也可以用于游标，尤其是在Rx版的客户端，可能更为方便。
connection.prepare("SELECT * FROM users WHERE age > @p1", ar1 -> {
  if (ar1.succeeded()) {
    PreparedStatement pq = ar1.result();
    // Fetch 50 rows at a time
    RowStream<Row> stream = pq.createStream(50, Tuple.of(18));
    // Use the stream
    stream.exceptionHandler(err -> {
      System.out.println("Error: " + err.getMessage());
    stream.endHandler(v -> {
      System.out.println("End of stream");
    stream.handler(row -> {
      System.out.println("User: " + row.getString("last_name"));
上边的stream会批量读取 50 行并同时将其转换为流，当这些行记录被传递给处理器时，
会以此类推地读取下一批的 50 行记录。
stream支持重启或暂停，已经加载到的行记录将会被保留在内存里直到被传递给处理器，此时
游标也将终止遍历。
client
  .preparedQuery("INSERT INTO colors VALUES (@p1)")
  .execute(Tuple.of(Color.red),  res -> {
    // ...
您可以像这样解码 Java 枚举：
client
  .preparedQuery("SELECT color FROM colors")
  .execute()
  .onComplete(res -> {
    if (res.succeeded()) {
      RowSet<Row> rows = res.result();
      for (Row row : rows) {
        System.out.println(row.get(Color.class, "color"));
  .addString(null);
client
  .preparedQuery("INSERT INTO movies (id, title, plot) VALUES (@p1, @p2, @p3)")
  .execute(tuple, res -> {
    // ...
否则，您应该使用 NullValue 常量和 NullValue.of 方法之一以显式声明类型：
Tuple tuple = Tuple.of(17, "The Man Who Knew Too Much", NullValue.String);
client
  .preparedQuery("INSERT INTO movies (id, title, plot) VALUES (@p1, @p2, @p3)")
  .execute(tuple, res -> {
    // ...
Collector<Row, ?, Map<Long, String>> collector = Collectors.toMap(
  row -> row.getLong("id"),
  row -> row.getString("last_name"));
// 使用收集器运行查询
client.query("SELECT * FROM users")
  .collecting(collector)
  .execute(ar -> {
    if (ar.succeeded()) {
      SqlResult<Map<Long, String>> result = ar.result();
      // 获取收集器创建的映射（map）对象
      Map<Long, String> map = result.value();
      System.out.println("Got " + map);
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
收集器处理时不得保留对 Row 的引用，
因为有一行用于处理整个集合。
Java Collectors 提供了许多有趣的预定义收集器，例如您可以
create 直接从行集轻松创建字符串：
Collector<Row, ?, String> collector = Collectors.mapping(
  row -> row.getString("last_name"),
  Collectors.joining(",", "(", ")")
// 使用收集器运行查询
client.query("SELECT * FROM users")
  .collecting(collector)
  .execute(ar -> {
    if (ar.succeeded()) {
      SqlResult<String> result = ar.result();
      // 获取收集器创建的字符串
      String list = result.value();
      System.out.println("Got " + list);
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
connection.infoHandler(info -> {
  System.out.println("Received info " + info.getSeverity() + "" + info.getMessage());
MSSQLPool pool = MSSQLPool.pool(database, new PoolOptions().setMaxSize(maxSize));
vertx.deployVerticle(() -> new AbstractVerticle() {
  @Override
  public void start() throws Exception {
    // 使用连接池
}, new DeploymentOptions().setInstances(4));
您也可以用以下方式在每个 Verticle 中创建可共享的连接池：
vertx.deployVerticle(() -> new AbstractVerticle() {
  MSSQLPool pool;
  @Override
  public void start() {
    // 创建一个可共享的连接池
    // 或获取已有的可共享连接池，并创建对原连接池的借用
    // 当 verticle 被取消部署时，借用会被自动释放
    pool = MSSQLPool.pool(database, new PoolOptions()
      .setMaxSize(maxSize)
      .setShared(true)
      .setName("my-pool"));
}, new DeploymentOptions().setInstances(4));
第一次创建可共享的连接池时，会创建新连接池所需的资源。之后再调用该创建方法时，会复用之前的连接池，并创建
对原有连接池的借用。当所有的借用都被关闭时，该连接池的资源也会被释放。
默认情况下，客户端需要创建一个 TCP 连接时，会复用当前的 event-loop 。 这个可共享的 HTTP 客户端会
以一种安全的模式，在使用它的 verticle 中随机选中一个 verticle，并使用它的 event-loop。
您可以手动设置一个客户端可以使用的 event-loop 的数量
MSSQLPool pool = MSSQLPool.pool(database, new PoolOptions()
  .setMaxSize(maxSize)
  .setShared(true)
  .setName("my-pool")
  .setEventLoopSize(4));
pool.connectHandler(conn -> {
  conn.query(sql).execute().onSuccess(res -> {
    //  将连接释放回连接池，以被该应用程序复用
    conn.close();