响应式 Microsoft SQL Server 客户端

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

dependencies {
 compile 'io.vertx:vertx-mssql-client:4.5.35'
}

MSSQLConnectOptions connectOptions = new MSSQLConnectOptions()
  .setPort(1433)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// 连接池配置
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// 创建客户端连接池
Pool client = MSSQLBuilder.pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .build();

// 简单查询
client
  .query("SELECT * FROM users WHERE id='julien'")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      RowSet<Row> 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);

// 创建连接池客户端
Pool client = MSSQLBuilder.pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

MSSQLConnectOptions connectOptions = new MSSQLConnectOptions()
  .setPort(1433)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// 连接池配置
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);
// 创建连接池客户端
Pool client = MSSQLBuilder.pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

pool.close();

MSSQLConnectOptions connectOptions = new MSSQLConnectOptions()
  .setPort(1433)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// 连接池配置
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// 创建连接池客户端
Pool client = MSSQLBuilder.pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

// 从连接池获取连接
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());
  }
});

options
  .setReconnectAttempts(2)
  .setReconnectInterval(1000);

client
  .query("SELECT * FROM users WHERE id='julien'")
  .execute()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> result = ar.result();
    System.out.println("Got " + result.size() + " rows ");
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("INSERT INTO movies (title) OUTPUT INSERTED.id VALUES (@p1)")
  .execute(Tuple.of("The Man Who Knew Too Much"))
  .onComplete(res -> {
    if (res.succeeded()) {
      Row row = res.result().iterator().next();
      System.out.println(row.getLong("id"));
    }
  });

connection
  .prepare("SELECT * FROM users WHERE first_name LIKE @p1")
  .onComplete(ar0 -> {
  if (ar0.succeeded()) {
    PreparedStatement pq = ar0.result();

    // 游标要求在事务内部运行
    connection
      .begin()
      .onComplete(ar1 -> {
      if (ar1.succeeded()) {
        Transaction tx = ar1.result();

        // 创建游标
        Cursor cursor = pq.cursor(Tuple.of("julien"));

        // 读取 50 行
        cursor.read(50).onComplete(ar2 -> {
          if (ar2.succeeded()) {
            RowSet<Row> rows = ar2.result();

            // 检查是否还有剩余数据行？
            if (cursor.hasMore()) {
              // 重复读取流程...
            } else {
              // 全部数据行读取完毕 - 提交事务
              tx.commit();
            }
          }
        });
      }
    });
  }
});

cursor
  .read(50)
  .onComplete(ar2 -> {
  if (ar2.succeeded()) {
    // 关闭游标
    cursor.close();
  }
});

connection
  .prepare("SELECT * FROM users WHERE first_name LIKE @p1")
  .onComplete(ar0 -> {
  if (ar0.succeeded()) {
    PreparedStatement ps = ar0.result();

    // 数据流要求在事务内部运行
    connection
      .begin()
      .onComplete(ar1 -> {
      if (ar1.succeeded()) {
        Transaction tx = ar1.result();

        // 每次抓取 50 行
        RowStream<Row> stream = pq.createStream(50, Tuple.of("julien"));

        // 使用数据流
        stream.exceptionHandler(err -> {
          System.out.println("Error: " + err.getMessage());
          ps.close();
        });
        stream.endHandler(v -> {
          // 关闭流以释放数据库中的游标资源
          stream
            .close()
            .onComplete(closed -> {
            tx.commit()
              .onComplete(committed -> {
                System.out.println("End of stream");
                ps.close();
            });
          });
        });
        stream.handler(row -> {
          System.out.println("User: " + row.getString("last_name"));
        });
      }
    });
  }
});

options.setTracingPolicy(TracingPolicy.ALWAYS);

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()
  .onComplete(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());
    }
  });

Collector<Row, ?, String> collector = Collectors.mapping(
  row -> row.getString("last_name"),
  Collectors.joining(",", "(", ")")
);

// 配合收集器执行查询
client.query("SELECT * FROM users")
  .collecting(collector)
  .execute()
  .onComplete(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());
});

Pool pool = Pool.pool(database, new PoolOptions().setMaxSize(maxSize));
vertx.deployVerticle(() -> new VerticleBase() {
  @Override
  public Future<?> start() throws Exception {
    // 使用连接池
    return super.start();
  }
}, new DeploymentOptions().setInstances(4));

vertx.deployVerticle(() -> new VerticleBase() {
  Pool pool;
  @Override
  public Future<?> start() throws Exception {
    // 获取或创建共享连接池
    // 本质上是获取对该池的一份租约（Lease）
    // 当该 Verticle 卸载时，租约会自动释放
    pool = MSSQLBuilder.pool()
      .with(new PoolOptions()
        .setMaxSize(maxSize)
        .setShared(true)
        .setName("my-pool"))
      .connectingTo(database)
      .using(vertx)
      .build();
    return super.start();
  }
}, new DeploymentOptions().setInstances(4));

Pool pool = MSSQLBuilder.pool()
  .with(new PoolOptions()
    .setMaxSize(maxSize)
    .setShared(true)
    .setName("my-pool")
    .setEventLoopSize(4))
  .connectingTo(database)
  .using(vertx)
  .build();