响应式 MySQL 客户端（Reactive MySQL Client）

MySQLConnectOptions connectOptions = new MySQLConnectOptions()
  .setHost("/var/run/mysqld/mysqld.sock")
  .setDatabase("the-db");

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

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

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

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

// 通过数据对象创建连接池
Pool pool = MySQLBuilder
  .pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

pool.getConnection()
  .onComplete(ar -> {
    // 处理你的连接
  });

String connectionUri = "mysql://dbuser:secretpassword@database.server.com:3306/mydb";

// 通过连接 URI 创建连接池
Pool pool = MySQLBuilder
  .pool()
  .connectingTo(connectionUri)
  .using(vertx)
  .build();

// 通过连接 URI 创建独立连接
MySQLConnection.connect(vertx, connectionUri)
  .onComplete(res -> {
    // 处理你的连接
  });

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

client
  .preparedQuery("SELECT first_name, last_name FROM users")
  .execute()
  .onComplete(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());
  }
});

client
  .preparedQuery("INSERT INTO users (first_name, last_name) VALUES (?, ?)")
  .execute(Tuple.of("Julien", "Viet"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    System.out.println(rows.rowCount());
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

System.out.println("User " + row.getString(0) + " " + row.getString(1));
System.out.println("User " + row.getString("first_name") + " " + row.getString("last_name"));

List<Tuple> batch = new ArrayList<>();
batch.add(Tuple.of("julien", "Julien Viet"));
batch.add(Tuple.of("emad", "Emad Alblueshi"));

// 执行批处理
client
  .preparedQuery("INSERT INTO USERS (id, name) VALUES (?, ?)")
  .executeBatch(batch)
  .onComplete(res -> {
  if (res.succeeded()) {

    // 获取所有行结果
    RowSet<Row> rows = res.result();
  } else {
    System.out.println("Batch failed " + res.cause());
  }
});

pool
  .getConnection()
  .compose(connection ->
    connection
      .preparedQuery("INSERT INTO Users (first_name,last_name) VALUES (?, ?)")
      .executeBatch(Arrays.asList(
        Tuple.of("Julien", "Viet"),
        Tuple.of("Emad", "Alblueshi")
      ))
      .compose(res -> connection
        // 处理返回的行数据
        .query("SELECT COUNT(*) FROM Users")
        .execute()
        .map(rows -> rows.iterator().next().getInteger(0)))
      // 将连接安全归还给连接池
      .eventually(() -> 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 ?")
  .compose(pq ->
    pq.query()
      .execute(Tuple.of("Julien"))
      .eventually(() -> pq.close())
  ).onSuccess(rows -> {
  // 获取所有行数据
});

pool.withConnection(connection ->
  connection
    .preparedQuery("INSERT INTO Users (first_name,last_name) VALUES (?, ?)")
    .executeBatch(Arrays.asList(
      Tuple.of("Julien", "Viet"),
      Tuple.of("Emad", "Alblueshi")
    ))
    .compose(res -> connection
      // 处理查询返回的行数据
      .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);
});

pool.getConnection()
  // 事务必须在独占的物理连接中运行
  .onSuccess(conn -> {
    // 开启事务
    conn.begin()
      .compose(tx -> conn
        // 执行一系列操作
        .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
        .execute()
        .compose(res2 -> conn
          .query("INSERT INTO Users (first_name,last_name) VALUES ('Emad','Alblueshi')")
          .execute())
        // 提交事务
        .compose(res3 -> tx.commit()))
      // 无论成功还是失败，均将连接归还至连接池
      .eventually(() -> conn.close())
      .onSuccess(v -> System.out.println("Transaction succeeded"))
      .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
  });

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 ('Emad','Alblueshi')")
    .execute()
    // 转换为业务提示消息
    .map("Users inserted")))
  .onSuccess(v -> System.out.println("Transaction succeeded"))
  .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));

client
  .preparedQuery("SELECT * FROM students WHERE updated_time = ?")
  .execute(Tuple.of(LocalTime.of(19, 10, 25)))
  .onComplete(ar -> {
    // 处理结果
  });
// 借助隐式类型转换，此方式同样可行
client
  .preparedQuery("SELECT * FROM students WHERE updated_time = ?")
  .execute(Tuple.of("19:10:25"))
  .onComplete(ar -> {
    // 处理结果
  });

client
  .query("SELECT graduated FROM students WHERE id = 0")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      RowSet<Row> rowSet = ar.result();
      for (Row row : rowSet) {
        int pos = row.getColumnIndex("graduated");
        Byte value = row.get(Byte.class, pos);
        Boolean graduated = row.getBoolean("graduated");
      }
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

client
  .preparedQuery("UPDATE students SET graduated = ? WHERE id = 0")
  .execute(Tuple.of(true))
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("Updated with the boolean value");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

Tuple tuple = Tuple.of(
  Tuple.JSON_NULL,
  new JsonObject().put("foo", "bar"),
  3);

// 获取 JSON 数据
Object value = tuple.getValue(0); // 预期为 JSON_NULL

//
value = tuple.get(JsonObject.class, 1); // 预期为 JSON 对象

//
value = tuple.get(Integer.class, 2); // 预期为 3
value = tuple.getInteger(2); // 预期为 3

Numeric numeric = row.get(Numeric.class, 0);
if (numeric.isNaN()) {
  // 处理 NaN
} else {
  BigDecimal value = numeric.bigDecimalValue();
}

client
  .preparedQuery("INSERT INTO colors VALUES (?)")
  .execute(Tuple.of(Color.red))
  .onComplete(res -> {
    // ...
  });

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"));
    }
  }
});

client
  .query("SELECT ST_AsText(g) FROM geom;")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // 获取 WKT 格式的空间数据
      RowSet<Row> result = ar.result();
      for (Row row : result) {
        String wktString = row.getString(0);
      }
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

client
  .query("SELECT ST_AsBinary(g) FROM geom;")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // 获取 WKB 格式的空间数据
      RowSet<Row> result = ar.result();
      for (Row row : result) {
        Buffer wkbValue = row.getBuffer(0);
      }
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

client
  .query("SELECT g FROM geom;")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // 将空间数据作为 Vert.x 数据对象（Data Object）获取
      RowSet<Row> result = ar.result();
      for (Row row : result) {
        Point point = row.get(Point.class, 0);
        System.out.println("Point x: " + point.getX());
        System.out.println("Point y: " + point.getY());
      }
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

Point point = new Point(0, 1.5, 1.5);
// 以 WKB 表现形式发送
client
  .preparedQuery("INSERT INTO geom VALUES (ST_GeomFromWKB(?))")
  .execute(Tuple.of(point))
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("Success");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

MySQLConnectOptions options = new MySQLConnectOptions()
  .setPort(3306)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
  .setAuthenticationPlugin(MySQLAuthenticationPlugin.MYSQL_NATIVE_PASSWORD); // 设置默认认证插件

MySQLConnectOptions options1 = new MySQLConnectOptions()
  .setPort(3306)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
  .setServerRsaPublicKeyPath("tls/files/public_key.pem"); // 通过公钥文件路径配置

MySQLConnectOptions options2 = new MySQLConnectOptions()
  .setPort(3306)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
  .setServerRsaPublicKeyValue(Buffer.buffer("-----BEGIN PUBLIC KEY-----\n" +
    "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA3yvG5s0qrV7jxVlp0sMj\n" +
    "xP0a6BuLKCMjb0o88hDsJ3xz7PpHNKazuEAfPxiRFVAV3edqfSiXoQw+lJf4haEG\n" +
    "HQe12Nfhs+UhcAeTKXRlZP/JNmI+BGoBduQ1rCId9bKYbXn4pvyS/a1ft7SwFkhx\n" +
    "aogCur7iIB0WUWvwkQ0fEj/Mlhw93lLVyx7hcGFq4FOAKFYr3A0xrHP1IdgnD8QZ\n" +
    "0fUbgGLWWLOossKrbUP5HWko1ghLPIbfmU6o890oj1ZWQewj1Rs9Er92/UDj/JXx\n" +
    "7ha1P+ZOgPBlV037KDQMS6cUh9vTablEHsMLhDZanymXzzjBkL+wH/b9cdL16LkQ\n" +
    "5QIDAQAB\n" +
    "-----END PUBLIC KEY-----\n")); // 通过公钥 Buffer 内容配置

connection.ping().onComplete(ar -> {
  System.out.println("The server has responded to the PING");
});

connection
  .resetConnection()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("Connection has been reset now");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

MySQLAuthOptions authenticationOptions = new MySQLAuthOptions()
  .setUser("newuser")
  .setPassword("newpassword")
  .setDatabase("newdatabase");
connection
  .changeUser(authenticationOptions)
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("User of current connection has been changed.");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

connection
  .specifySchema("newschema")
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("Default schema changed to newschema");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

connection
  .getInternalStatistics()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("Statistics: " + ar.result());
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

connection
  .debug()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("Debug info dumped to server's STDOUT");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

connection
  .setOption(MySQLSetOption.MYSQL_OPTION_MULTI_STATEMENTS_OFF)
  .onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("CLIENT_MULTI_STATEMENTS is off now");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

Pool pool = MySQLBuilder.pool()
  .with(options)
  .connectingTo(Arrays.asList(server1, server2, server3))
  .using(vertx)
  .build();

builder.withConnectHandler(conn -> {
  conn.query(sql).execute().onSuccess(res -> {
    // 将连接归还给池，准备供应用程序正常借用
    conn.close();
  });
});

Pool pool = MySQLBuilder.pool()
  .with(poolOptions)
  .connectingTo(() -> {
    Future<SqlConnectOptions> connectOptions = retrieveOptions();
    return connectOptions;
  })
  .using(vertx)
  .build();