数据库与持久化
API 文档
你当前正在浏览历史版本 Vert.x 4.2.7 的文档。 访问本页面的 最新版本。

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

Reactive MySQL Client 是一款面向 MySQL 的客户端，具有简单易用的 API，专注于高伸缩性与低开销。

该客户端是响应式且非阻塞的，允许通过单个线程处理大量数据库连接。

特性

  • 事件驱动

  • 轻量级

  • 内置连接池

  • 预编译查询缓存

  • 游标（Cursor）支持

  • 行流式处理（Row streaming）

  • RxJava API

  • 内存直接映射到对象，无多余复制开销

  • 完整的数据类型支持

  • 存储过程支持

  • TLS/SSL 支持

  • 查询流水线（Query pipelining）

  • MySQL 实用命令支持

  • 同时支持 MySQL 与 MariaDB

  • 丰富的排序规则（Collation）与字符集支持

  • Unix 域套接字（Unix Domain Socket）

使用方法

要使用 Reactive MySQL Client，请将以下依赖项添加至构建描述文件的 dependencies 配置节中：

  • Maven（在 pom.xml 中）：

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

  • Gradle（在 build.gradle 文件中）：

dependencies {
  compile 'io.vertx:vertx-mysql-client:5.0.12'
}

快速入门

以下是建立连接、执行查询以及断开连接的最简单方式：

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

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

// 创建客户端连接池
SqlClient client = MySQLBuilder
  .client()
  .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();
  });

连接至 MySQL 数据库

大多数情况下，你将使用连接池来连接 MySQL：

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

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

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

池化客户端基于连接池工作，任何操作都会从连接池中借用一条连接以执行该操作，并在执行完毕后将其归还至连接池。

如果你是在 Vert.x 环境中运行，可以传入你的 Vertx 实例：

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

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

当不再需要连接池时，你必须将其关闭：

pool.close();

当你需要在同一条连接上执行多项操作时，需要使用客户端 连接（Connection）。

你可以很方便地从连接池中获取一条连接：

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

一旦你使用完毕该连接，必须关闭它以将其释放回连接池，从而能够被复用。

命令流水线（Command Pipelining）

在某些使用场景下，命令流水线可以提升数据库访问性能。

你可以将客户端配置为使用流水线：

Pool pool = MySQLBuilder.pool()
  .with(poolOptions)
  .connectingTo(connectOptions.setPipeliningLimit(16))
  .using(vertx)
  .build();

默认的流水线限制为 1，即禁用流水线。

如果你使用的代理不支持流水线，请不要启用流水线功能。 否则，代理可能会异常断开客户端连接。

连接池（Pool）与池化客户端（Pooled Client）

通过 MySQLBuilder，你可以创建一个连接池或池化客户端：

connectOptions.setPipeliningLimit(64);
SqlClient client = MySQLBuilder.client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

// 启用流水线
Future<RowSet<Row>> res1 = client.query(sql).execute();

// 连接池
Pool pool = MySQLBuilder.pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

// 未启用流水线
Future<RowSet<Row>> res2 = pool.query(sql).execute();

  • 连接池上的操作不会以流水线方式执行，仅当从连接池中显式获取的连接才支持流水线

  • 池化客户端上的操作默认支持流水线，但你无法从池化客户端中显式获取单条连接

连接池共享

你可以在多个 Verticle 或同一 Verticle 的多个实例之间共享同一个连接池。此类连接池应在 Verticle 外部创建，否则当创建它的 Verticle 被卸载时，连接池也会被随之关闭：

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

你也可以直接在各个 Verticle 内部声明获取共享连接池：

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

首次创建共享连接池时，系统会为该连接池分配底层资源。随后的调用将直接复用该池并创建新的租约。当所有租约均被关闭后，底层系统资源才会被完全释放。

默认情况下，当连接池需要创建 TCP 物理连接时，会复用当前的 EventLoop。因此共享连接池会随机使用正在使用它的各 Verticle 的 EventLoop。

你也可以独立于调用上下文，为连接池显式分配指定数量的 EventLoop：

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

Unix 域套接字（Unix Domain Socket）

出于简化配置、安全性或性能等原因，有时需要通过 Unix 域套接字（Unix Domain Socket） 进行连接。

为此，请在 MySQLConnectOptions#setHost 中设置域套接字的路径：

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();
当在 JDK 16+ 上运行或使用 原生传输（Native Transport） 时支持 Unix 域套接字。

配置

你可以通过多种方式来配置客户端。

数据对象（Data Object）

配置客户端的一种简单方式是指定一个 MySQLConnectOptions 数据对象。

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 -> {
    // 处理你的连接
  });

排序规则与字符集（Collations and Character Sets）

Reactive MySQL Client 支持配置排序规则（Collation）或字符集（Character Set），并将它们映射到相应的 java.nio.charset.Charset。 例如，你可以像这样为连接指定字符集：

MySQLConnectOptions connectOptions = new MySQLConnectOptions();

// 将连接字符集设置为 utf8，而非默认的 utf8mb4
connectOptions.setCharset("utf8");

Reactive MySQL Client 默认采用 utf8mb4 字符集。诸如密码和错误信息等字符串值始终使用 UTF-8 字符集进行解码。

characterEncoding 选项用于决定使用哪种 Java 字符集来编码字符串值（如查询字符串和参数值）。该字符集默认是 UTF-8，若设置为 null，则客户端将改用 Java 默认字符集。

你还可以像这样为连接指定排序规则（Collation）：

MySQLConnectOptions connectOptions = new MySQLConnectOptions();

// 将连接排序规则设置为 utf8_general_ci，而非默认的 utf8mb4_general_ci
// 设置排序规则会覆盖 charset 选项
connectOptions.setCharset("gbk");
connectOptions.setCollation("utf8_general_ci");

注意：在数据对象上设置排序规则会覆盖 charset 与 characterEncoding 选项。

你可以执行 SQL SHOW COLLATION; 或 SHOW CHARACTER SET; 来获取服务端支持的排序规则与字符集。

有关 MySQL 字符集与排序规则的更多信息，请参阅 MySQL 参考手册。

连接属性（Connection Attributes）

你还可以通过 setProperties 或 addProperty 方法配置连接属性。注意 setProperties 将覆盖默认的客户端属性。

MySQLConnectOptions connectOptions = new MySQLConnectOptions();

// 添加一个连接属性
connectOptions.addProperty("_java_version", "1.8.0_212");

// 覆盖属性
Map<String, String> attributes = new HashMap<>();
attributes.put("_client_name", "myapp");
attributes.put("_client_version", "1.0.0");
connectOptions.setProperties(attributes);

有关客户端连接属性的更多信息，请参阅 MySQL 参考手册。

useAffectedRows

你可以配置 useAffectedRows 选项，决定在连接到服务器时是否设置 CLIENT_FOUND_ROWS 标志。如果指定了 CLIENT_FOUND_ROWS 标志，则受影响行数（affected rows count）为匹配找到的行数数值，而不是实际被变更的行数。

有关此选项的更多信息，请参阅 MySQL 参考手册。

连接 URI

除了使用 MySQLConnectOptions 数据对象进行配置之外，若你想使用连接 URI 进行配置，我们也提供了另一种连接方式：

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 -> {
    // 处理你的连接
  });
在连接 URI 中配置参数将覆盖默认属性。

有关连接字符串格式的更多信息，请参阅 MySQL 参考手册。

当前客户端支持以下参数键（不区分大小写）：

  • host

  • port

  • user

  • password

  • schema（数据库）

  • useaffectedrows

其余附加参数将被添加至 properties 中。

连接 URI 解析器会将所有属性键转换为小写。

连接重试

你可以将客户端配置为在建立连接失败时自动重试。

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

预编译查询（Prepared Query）

你同样可以对预编译查询执行类似操作。

SQL 字符串中可以通过位置引用参数，使用数据库特定的占位符语法 `?`：

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

查询方法返回异步的 RowSet 实例，适用于各类 SELECT 查询：

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

Row 允许通过索引序号或列名访问数据字段：

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

客户端不会对参数类型进行任何魔法猜测，所有的参数都应当严格遵守 SQL 类型系统。

你可以执行批处理预编译查询：

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

MySQL LAST_INSERT_ID

向数据表中插入记录后，你可以获取自增主键的值。

client
  .query("INSERT INTO test(val) VALUES ('v1')")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      RowSet<Row> rows = ar.result();
      long lastInsertId = rows.property(MySQLClient.LAST_INSERTED_ID);
      System.out.println("Last inserted id is: " + lastInsertId);
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

更多信息可参阅 How to Get the Unique ID for the Last Inserted Row。

使用独立连接

获取数据库连接

当你需要执行一系列顺序查询（不包含事务）时，可以创建新连接或直接从连接池借取连接。请牢记，在从池中获取连接到归还连接期间，你需要妥善维护该连接，因为它可能因服务器端空闲超时等原因被意外关闭：

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

同样可以创建并使用预编译查询（Prepared Query）：

connection
  .prepare("SELECT * FROM users WHERE first_name LIKE ?")
  .compose(pq ->
    pq.query()
      .execute(Tuple.of("Julien"))
      .eventually(() -> pq.close())
  ).onSuccess(rows -> {
  // 获取所有行数据
});

简化的连接管理 API

在使用连接池时，可以调用 withConnection 方法，传入一个在连接作用域内执行的函数。

该方法会自动从连接池借出连接，并将该连接作为参数调用你的函数。

传入的函数必须返回一个包含任意结果的 Future。

在该 Future 完成后，连接会被自动安全归还给连接池，并对外返回最终的处理结果：

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

使用事务

基于连接的事务管理

你可以使用 SQL BEGIN/COMMIT/ROLLBACK 语句手动执行事务；若这么做，你必须获取并自行管理 SqlConnection 连接。

或者，推荐使用 SqlConnection 提供的事务 API：

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

当数据库服务端判定当前事务失败时（例如出现著名的 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");
  });

简化的事务管理 API

在使用连接池时，你可以调用 withTransaction，传入一个在事务内执行的函数。

该方法会自动从连接池借取连接、开启事务，并调用你的业务函数，由客户端在该事务的上下文中执行所有数据库操作。

该业务函数必须返回包含任意操作结果的 Future：

  • 当 Future 成功时，客户端会自动提交该事务；

  • 当 Future 失败时，客户端会自动回滚该事务。

事务执行完毕（无论提交或回滚）后，连接都会自动归还给连接池，并向调用方返回整体操作的最终结果：

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()));

游标（Cursor）与流式查询

默认情况下，执行预编译查询会一次性抓取所有数据行。你可以使用 Cursor 来精准控制每次读取的数据行数：

connection
  .prepare("SELECT * FROM users WHERE first_name LIKE ?")
  .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();
  }
});

游标还提供了更方便的流式 API，尤其是在搭配 RxJava 版本时极为顺畅：

connection
  .prepare("SELECT * FROM users WHERE first_name LIKE ?")
  .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"));
        });
      }
    });
  }
});

数据流会以每次 50 行的批次读取并推送；当该批数据行全部传递给处理器后，再读取新的 50 行批次，以此类推。

该流支持随时暂停（Pause）或恢复（Resume）；已加载的行将缓存在内存中直到被消费，在此期间游标会暂停迭代抓取。

追踪查询（Tracing Queries）

当 Vert.x 启用了链路追踪（Tracing）时，SQL 客户端可以对查询执行过程进行全链路追踪。

客户端会上报以下 客户端（Client） Span：

  • 操作名称为 Query

  • 包含以下 Tags：

  • db.system：数据库管理系统产品名称

  • db.user：数据库用户名

  • db.instance：数据库实例

  • db.statement：SQL 查询语句

  • db.type：固定为 sql

默认的链路追踪策略为 PROPAGATE，客户端仅在参与现有活跃链路时才会创建 Span。

你可以通过 setTracingPolicy 更改客户端追踪策略，例如设置为 ALWAYS 以始终上报 Span：

options.setTracingPolicy(TracingPolicy.ALWAYS);

MySQL 类型映射

当前客户端支持以下 MySQL 类型：

MySQL 类型 Java 类型 备注

BOOLEAN / BOOL

java.lang.Byte

TINYINT(1) 的同义词

TINYINT

java.lang.Byte

TINYINT UNSIGNED

java.lang.Short

SMALLINT

java.lang.Short

SMALLINT UNSIGNED

java.lang.Integer

MEDIUMINT

java.lang.Integer

MEDIUMINT UNSIGNED

java.lang.Integer

INT / INTEGER

java.lang.Integer

INTEGER UNSIGNED

java.lang.Long

BIGINT

java.lang.Long

BIGINT UNSIGNED

io.vertx.sqlclient.data.Numeric

Java 中无无符号 64 位整型

FLOAT

java.lang.Float

FLOAT UNSIGNED

java.lang.Float

DOUBLE

java.lang.Double

DOUBLE UNSIGNED

java.lang.Double

BIT

java.lang.Long

NUMERIC / DECIMAL

io.vertx.sqlclient.data.Numeric

NUMERIC UNSIGNED

io.vertx.sqlclient.data.Numeric

DATE

java.time.LocalDate

DATETIME

java.time.LocalDateTime

TIME

java.time.Duration

亦可通过 java.time.LocalTime 检索

TIMESTAMP

java.time.LocalDateTime

YEAR

java.lang.Short

CHAR

java.lang.String

VARCHAR

java.lang.String

BINARY

io.vertx.core.buffer.Buffer

VARBINARY

io.vertx.core.buffer.Buffer

TINYBLOB

io.vertx.core.buffer.Buffer

TINYTEXT

java.lang.String

BLOB

io.vertx.core.buffer.Buffer

TEXT

java.lang.String

MEDIUMBLOB

io.vertx.core.buffer.Buffer

MEDIUMTEXT

java.lang.String

LONGBLOB

io.vertx.core.buffer.Buffer

LONGTEXT

java.lang.String

ENUM

java.lang.String

SET

java.lang.String

JSON

java.lang.Object

参见 处理 JSON 类型

GEOMETRY

io.vertx.mysqlclient.data.spatial.*

参见 处理 GEOMETRY 类型

Tuple（元组）解码在存储值时会使用上述类型。

Java 原生没有专门针对无符号数值的表示类型。客户端会将无符号值转换为足以容纳该数值的对应 Java 类型。

隐式类型转换

Reactive MySQL Client 支持在执行预编译语句（Prepared Statement）时进行隐式类型转换。 假设你的表中包含一个 TIME 列，以下两个示例均可正常执行：

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 -> {
    // 处理结果
  });

编码使用的 MySQL 数据类型将根据参数值推断，类型映射如下：

参数值类型 编码对应的 MySQL 类型

null

MYSQL_TYPE_NULL

java.lang.Byte

MYSQL_TYPE_TINY

java.lang.Boolean

MYSQL_TYPE_TINY

java.lang.Short

MYSQL_TYPE_SHORT

java.lang.Integer

MYSQL_TYPE_LONG

java.lang.Long

MYSQL_TYPE_LONGLONG

java.lang.Double

MYSQL_TYPE_DOUBLE

java.lang.Float

MYSQL_TYPE_FLOAT

java.time.LocalDate

MYSQL_TYPE_DATE

java.time.Duration

MYSQL_TYPE_TIME

java.time.LocalTime

MYSQL_TYPE_TIME

io.vertx.core.buffer.Buffer

MYSQL_TYPE_BLOB

java.time.LocalDateTime

MYSQL_TYPE_DATETIME

io.vertx.mysqlclient.data.spatial.*

MYSQL_TYPE_BLOB

default

MYSQL_TYPE_STRING

处理 BOOLEAN 类型

在 MySQL 中，BOOLEAN 与 BOOL 数据类型是 TINYINT(1) 的同义词。数值 0 被视作 false，非零值被视作 true。 BOOLEAN 数据类型的值在 Row 或 Tuple 中以 java.lang.Byte 类型存储。你可以调用 Row#getValue 将其作为 java.lang.Byte 获取，也可以调用 Row#getBoolean 将其作为 java.lang.Boolean 获取。

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

当你需要使用 BOOLEAN 类型的参数执行预编译语句时，只需直接将 java.lang.Boolean 值添加至参数列表中即可。

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

处理 JSON 类型

MySQL JSON 数据类型由以下 Java 类型表示：

  • String

  • Number

  • Boolean

  • io.vertx.core.json.JsonObject

  • io.vertx.core.json.JsonArray

  • io.vertx.sqlclient.Tuple#JSON_NULL 用于表示 JSON 的 null 字面量

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

处理 BIT 类型

BIT 数据类型映射至 java.lang.Long 类型，但 Java 本身没有无符号数值概念，因此如果你想插入或更新 BIT(64) 的最大值记录，可以通过技巧将参数设置为 -1L。

处理 TIME 类型

MySQL 的 TIME 数据类型既可以表示一天中的某个时间，也可以表示时间间隔（范围从 -838:59:59 到 838:59:59）。在 Reactive MySQL Client 中，TIME 数据类型原生映射至 java.time.Duration，但你也可以通过 Row#getLocalTime 访问器将其作为 java.time.LocalTime 获取。

处理 NUMERIC 类型

Numeric Java 类型用于表示 MySQL 的 NUMERIC 类型。

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

处理 ENUM 类型

MySQL 支持 ENUM 数据类型，客户端会将这些类型作为 String 数据类型获取。

你可以像这样将 Java 枚举编码为 String：

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

你可以像这样将 ENUM 列作为 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"));
    }
  }
});

处理 GEOMETRY 类型

客户端还支持 MySQL 的 GEOMETRY 数据类型。以下示例展示了你可以按知名文本（WKT, Well-Known Text）格式或知名二进制（WKB, Well-Known Binary）格式处理几何数据，这些数据将被解码为 MySQL TEXT 或 BLOB 数据类型。业界有许多处理此类格式数据的优秀第三方库。

你可以获取 WKT 格式的空间数据：

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

或者获取 WKB 格式的空间数据：

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

在 Reactive MySQL Client 中，我们还提供了一种处理几何数据类型的简便方式。

你可以将几何数据作为 Vert.x 数据对象（Data Object）获取：

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

你也可以将其作为 WKB 表示的预编译语句参数传入：

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

收集器查询（Collector Queries）

你可以在查询 API 中结合使用 Java 收集器（Collector）：

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

收集器处理逻辑中绝对不能保留对 Row 的引用，因为处理整个结果集时复用的是同一个 Row 对象。

Java Collectors 提供了许多实用的预定义收集器，例如你可以轻松地直接从行结果集中拼装出一个字符串：

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

MySQL 存储过程

你可以在查询中执行存储过程。结果将遵循 MySQL 协议 从服务端直接获取，没有任何特殊魔法。

client.query("CREATE PROCEDURE multi() BEGIN\n" +
  "  SELECT 1;\n" +
  "  SELECT 1;\n" +
  "  INSERT INTO ins VALUES (1);\n" +
  "  INSERT INTO ins VALUES (2);\n" +
  "END;")
  .execute()
  .onComplete(ar1 -> {
  if (ar1.succeeded()) {
    // 成功创建存储过程
    client
      .query("CALL multi();")
      .execute()
      .onComplete(ar2 -> {
      if (ar2.succeeded()) {
        // 处理结果
        RowSet<Row> result1 = ar2.result();
        Row row1 = result1.iterator().next();
        System.out.println("First result: " + row1.getInteger(0));

        RowSet<Row> result2 = result1.next();
        Row row2 = result2.iterator().next();
        System.out.println("Second result: " + row2.getInteger(0));

        RowSet<Row> result3 = result2.next();
        System.out.println("Affected rows: " + result3.rowCount());
      } else {
        System.out.println("Failure: " + ar2.cause().getMessage());
      }
    });
  } else {
    System.out.println("Failure: " + ar1.cause().getMessage());
  }
});

注意：目前暂不支持预编译语句绑定 OUT 输出参数。

MySQL LOCAL INFILE

本客户端支持处理 LOCAL INFILE 请求。如果你想将本地文件中的数据加载到服务器表中，可以使用查询： LOAD DATA LOCAL INFILE '<filename>' INTO TABLE <table>;。更多信息可参阅 MySQL 参考手册。

身份认证

默认认证插件

本客户端支持在连接建立之初指定所使用的默认认证插件。目前支持以下插件：

  • mysql_native_password

  • caching_sha2_password

  • mysql_clear_password

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

MySQL 8 引入的新认证方式

MySQL 8.0 引入了一种名为 caching_sha2_password 的全新身份认证方式，并将其设为默认认证机制。 为了使用这种全新的认证方式连接到服务端，你需要使用安全连接（即启用 TLS/SSL），或者使用 RSA 密钥对交换加密密码以防止密码泄漏。RSA 密钥对会在通信过程中自动交换，但由于在非安全连接上传输，服务端的 RSA 公钥在此过程中可能面临被中间人劫持篡改的风险。 因此，如果你处于非安全连接中，且希望规避暴露服务端 RSA 公钥的风险，可以像这样显式配置服务端 RSA 公钥：

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 内容配置

有关 caching_sha2_password 认证方式的更多信息，请参阅 MySQL 参考手册。

使用 SSL/TLS

要将客户端配置为使用 SSL 连接，你可以像配置 Vert.x NetClient 一样配置 MySQLConnectOptions。

客户端支持所有的 SSL 模式，你可以通过 sslmode 进行配置。 客户端默认处于 PREFERRED（偏好）SSL 模式：当服务端支持加密时会自动升级为 SSL 连接，否则降级为非 SSL 连接。 这与 MySQL Connector/J 的行为保持一致。

若要显式禁用 SSL，请在连接选项中将 SSL 模式设置为 DISABLED。 
MySQLConnectOptions options = new MySQLConnectOptions()
  .setPort(3306)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
  .setSslMode(SslMode.VERIFY_CA)
  .setSslOptions(new ClientSSLOptions().setTrustOptions(new PemTrustOptions().addCertPath("/path/to/cert.pem")));

MySQLConnection.connect(vertx, options)
  .onComplete(res -> {
    if (res.succeeded()) {
      // 已通过 SSL 成功连接
    } else {
      System.out.println("Could not connect " + res.cause());
    }
  });

更多信息请参阅 Vert.x 文档。

MySQL 实用命令（Utility Commands）

有时你可能需要使用 MySQL 实用命令，客户端对此提供了相应支持。 更多信息可参阅 MySQL 实用命令。

COM_PING

你可以使用 COM_PING 命令检查服务端是否存活。如果服务端响应了 PING 请求，处理器（Handler）将收到通知；否则该处理器永远不会被调用。

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

COM_RESET_CONNECTION

你可以使用 COM_RESET_CONNECTION 命令重置会话状态，这将重置如下连接状态： - 用户变量 - 临时表 - 预编译语句

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

COM_CHANGE_USER

你可以更改当前连接的用户，这将执行重新认证并像 COM_RESET_CONNECTION 一样重置连接状态。

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

COM_INIT_DB

你可以使用 COM_INIT_DB 命令更改连接的默认 Schema（数据库）。

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

COM_STATISTICS

你可以使用 COM_STATISTICS 命令获取 MySQL 服务端某些内部状态变量的人类可读字符串。

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

COM_DEBUG

你可以使用 COM_DEBUG 命令将调试信息转储到 MySQL 服务端的标准输出（STDOUT）。

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

COM_SET_OPTION

你可以使用 COM_SET_OPTION 命令为当前连接设置选项。目前仅支持设置 CLIENT_MULTI_STATEMENTS。

例如，你可以通过该命令禁用 CLIENT_MULTI_STATEMENTS：

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

MySQL 与 MariaDB 版本支持矩阵

MySQL MariaDB

版本

是否支持

版本

是否支持

5.5

✔

10.1

✔

5.6

✔

10.2

✔

5.7

✔

10.3

✔

8.0

✔

10.4

✔

已知问题：

  • 重置连接实用命令（Reset connection）在 MySQL 5.5、5.6 以及 MariaDB 10.1 中无法使用

  • 变更用户实用命令（Change user）在 MariaDB 10.2 和 10.3 中不受支持

常见陷阱与最佳实践

以下是一些最佳实践，可帮助你在使用 Reactive MySQL Client 时规避常见陷阱。

预编译语句数量限制

有时你可能会遇到臭名昭著的错误：Can’t create more than max_prepared_stmt_count statements (current value: 16382)，这是由于服务端达到了预编译语句总数的上限所致。

你可以调整服务端的系统变量 max_prepared_stmt_count，但该变量存在上限值，因此无法通过这种方式彻底消除该错误。

缓解该问题的最佳方式是启用预编译语句缓存（Prepared Statement Caching），这样拥有相同 SQL 字符串的预编译语句就能得以复用，客户端无需为每个请求都重新创建全新的预编译语句。 在语句执行完毕后，预编译语句将被自动关闭。 通过这种方式，达到限制上限的概率将大幅降低，尽管无法完全杜绝。

你也可以通过 SqlConnection#prepare 接口创建 PreparedStatement 对象来手动管理预编译语句的生命周期，从而自主掌控何时释放语句句柄，甚至可以使用 SQL 语法级别的预编译语句。

揭秘预编译批处理（Prepared Batch）

当你需要向数据库批量插入数据时，可以使用 PreparedQuery#executeBatch，它提供了简洁的 API 来处理该场景。 但请牢记：MySQL 原生协议并不支持真正的批量处理，因此该 API 本质上只是一种语法糖——通过依次逐个执行预编译语句实现。这意味着与单次执行带有值列表的预编译语句插入多行相比，该方式需要更多的网络往返（RTT）。

棘手的 DATE 与 TIME 数据类型

处理 MySQL 的 DATE 和 TIME 数据类型（尤其是涉及不同时区时）颇为棘手，因此 Reactive MySQL Client 不会对这些值进行任何隐式魔法转换。

  • MySQL 的 DATETIME 数据类型不包含时区信息，因此无论当前会话处于何种时区，你取出的值都与你存入的值完全一致。

  • MySQL 的 TIMESTAMP 数据类型包含时区信息，因此在存入或取出值时，服务端始终会根据当前会话中设置的时区对其进行转换。

高级连接池配置

服务器负载均衡

你可以为连接池配置多个服务器地址列表以实现负载均衡，而不是仅指定单一服务器：

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

连接池在创建底层物理连接时采用轮询（Round-Robin）算法在不同服务器之间进行分配。

该负载均衡策略发生在连接创建阶段，而非从池中借取（Borrow）连接的阶段。

连接池初始化处理器

你可以使用 withConnectHandler 在物理连接创建完成后、放入连接池之前与其进行交互并执行初始化设置：

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

一旦你完成对该连接的初始化操作，只需直接调用 close() 将其标记为就绪，以便连接池开始调度使用。

动态连接配置

你可以使用 Java Supplier 函数式接口动态提供连接配置详情，而非传入静态的 SqlConnectOptions 实例。

由于该 Supplier 是异步的，因此可用于实现连接配置的动态变更（例如数据库密码自动轮换）：

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