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

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

Reactive PostgreSQL Client 是一款针对 PostgreSQL 的客户端，提供直观易用的 API，专注于高伸缩性与低开销。

该客户端采用响应式与非阻塞设计，允许仅凭单个线程即可处理大量数据库连接。

  • 特性列表

    • 事件驱动

    • 轻量高效

    • 内置连接池

    • 预编译查询（Prepared queries）缓存

    • 基于 PostgreSQL NOTIFY/LISTEN 的发布 / 订阅

    • 批量操作与游标

    • 行流式处理（Row streaming）

    • 命令流水线（Command pipelining）

    • RxJava API

    • 零多余拷贝，直接从内存映射到对象

    • 支持 Java 8 日期和时间类型

    • 支持 SSL/TLS

    • 支持 Unix Domain Socket

    • 支持 HTTP/1.x CONNECT、SOCKS4a 或 SOCKS5 代理

    • 支持 4 层与 7 层代理（Proxy）

引入依赖（Usage）

要使用 Reactive PostgreSQL Client，请将以下依赖项添加到构建描述文件的 dependencies 部分：

  • Maven（在 pom.xml 中）：

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

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

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

快速入门（Getting started）

以下是连接、查询以及断开连接的最简单示例：

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

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

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

连接至 PostgreSQL 数据库

大多数情况下，你会使用连接池来连接 PostgreSQL：

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

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

// 创建池化客户端
SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .build();

池化客户端底层使用连接池，执行任何操作时都会从连接池借取连接执行该操作，执行完毕后将其归还给连接池。

如果你在 Vert.x 环境中运行，可以将你的 Vertx 实例传递给它：

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

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

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

当你不再需要该客户端时，应当将其释放关闭：

client.close();

当你需要在同一连接上连续执行多个操作时，需要从连接池中获取一个 连接（connection）。

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

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

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

// 创建池化客户端
Pool pool = PgBuilder
  .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 = PgBuilder
  .pool()
  .connectingTo(connectOptions.setPipeliningLimit(16))
  .with(poolOptions)
  .using(vertx)
  .build();

默认的流水线限制（pipelining limit）为 256。

你可以将该值设置为 1 以禁用流水线。

连接池 vs 池化客户端（Pool versus pooled client）

PgBuilder 允许你创建连接池（pool）或池化客户端（pooled client）：

SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

// 流水线式执行
Future<RowSet<Row>> res1 = client.query(sql).execute();

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

// 非流水线式执行
Future<RowSet<Row>> res2 = pool.query(sql).execute();

  • 连接池（pool）操作默认不采用流水线式，只有从连接池借出的连接才会进行流水线式处理

  • 池化客户端（pooled client）操作采用流水线式，但你无法直接从池化客户端中借取连接

连接池共享（Pool sharing）

Unix Domain Socket

有时出于易用性、安全性或性能方面的考虑，需要通过 Unix Domain Socket 进行连接。

为此，请在 PgConnectOptions#setHost 中指定 domain socket 的文件路径：

PgConnectOptions connectOptions = new PgConnectOptions()
  .setHost("/var/run/postgresql")
  .setPort(5432)
  .setDatabase("the-db");

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

// 创建池化客户端
Pool client = PgBuilder
  .pool()
  .connectingTo(connectOptions)
  .with(poolOptions)
  .using(vertx)
  .build();
当运行在 JDK 16+ 上，或使用 本地传输（native transport） 时，支持 Unix Domain Socket。

连接重试（Connect retries）

你可以配置客户端在建立连接失败时自动进行重试：

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

配置（Configuration）

客户端为你提供了多种配置方式。

数据对象（data object）

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

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

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

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

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

你也可以通过 setProperties 或 addProperty 方法配置通用属性。请注意，setProperties 会覆盖客户端的默认属性。

当将此客户端与 CockroachDB DBaaS 一起使用时，需要通过 addProperty("options", "--cluster=<cluster-id>") 或在 URL 中使用 …​&options=--cluster%3D<cluster-id> 包含 cluster 选项。

例如，你可以通过添加 search_path 属性为连接设置默认 schema：

PgConnectOptions connectOptions = new PgConnectOptions();

// 设置默认 schema
Map<String, String> props = new HashMap<>();
props.put("search_path", "myschema");
connectOptions.setProperties(props);

有关可用属性的更多信息，请参阅 PostgreSQL 官方手册。

连接 URI（Connection URI）

除了使用 PgConnectOptions 数据对象进行配置外，如果你想通过连接 URI 进行配置，我们也为你提供了另一种连接方式：

String connectionUri = "postgresql://dbuser:secretpassword@database.server.com:5432/mydb";

// 基于连接 URI 创建连接池
Pool pool = PgBuilder.pool()
  .connectingTo(connectionUri)
  .using(vertx)
  .build();

// 基于连接 URI 创建连接
PgConnection
  .connect(vertx, connectionUri)
  .onComplete(res -> {
    // 处理你的数据库连接
  });
在连接 URI 中配置的参数将会覆盖默认属性。

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

目前，客户端支持以下参数键：

  • host

  • hostaddr

  • port

  • user

  • password

  • dbname

  • sslmode

  • sslnegotiation

其他额外参数将被添加至 properties 中。

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

环境变量（environment variables）

你也可以使用环境变量来设置默认连接配置值，这在需要避免硬编码数据库连接信息时非常实用。你可以参阅 官方文档 获取更多详情。支持以下环境变量：

  • PGHOST

  • PGHOSTADDR

  • PGPORT

  • PGDATABASE

  • PGUSER

  • PGPASSWORD

  • PGSSLMODE

  • PGSSLNEGOTIATION

如果你在连接时未指定数据对象或连接 URI 字符串，环境变量将具有更高的优先级。

$ PGUSER=user \
  PGHOST=the-host \
  PGPASSWORD=secret \
  PGDATABASE=the-db \
  PGPORT=5432 \
  PGSSLMODE=DISABLE
Pool pool = PgBuilder.pool()
  .using(vertx)
  .build();

// 从环境变量创建连接
PgConnection.connect(vertx)
  .onComplete(res -> {
    // 处理你的数据库连接
  });

SASL SCRAM-SHA-256 认证机制

vertx-pg-client 支持 SCRAM-SHA-256 和 SCRAM-SHA-256-PLUS SASL 认证机制。从 v5.0.0 版本开始，vertx-pg-client 自带了 com.ongres.scram 库依赖，从旧版本迁移的用户可以从其 pom.xml 和 build.gradle 文件中移除 com.ongres.scram 库。

如果客户端仅使用其他认证方式，则可以使用 Maven <exclusions> 或 Gradle exclude 排除传递性 com.ongres.scram 依赖，以减小软件打包体积。

执行查询

当你不需要事务或仅运行单次查询时，可以直接在连接池上执行查询；连接池会自动从池中借调一个连接来执行该查询并向你返回结果。

以下演示如何执行简单查询：

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 字符串中可以通过位置引用参数，使用数据库特定的占位符语法 `$1`, `$2`, etc…​：

client
  .preparedQuery("SELECT * FROM users WHERE id=$1")
  .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 ($1, $2)")
  .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 ($1, $2)")
  .executeBatch(batch)
  .onComplete(res -> {
  if (res.succeeded()) {

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

RETURNING 子句

你可以在查询中使用 'RETURNING' 子句获取自动生成的主键：

client
  .preparedQuery("INSERT INTO color (color_name) VALUES ($1), ($2), ($3) RETURNING color_id")
  .execute(Tuple.of("white", "red", "blue"))
  .onSuccess(rows -> {
    for (Row row : rows) {
      System.out.println("generated key: " + row.getInteger("color_id"));
    }
});

只要 SQL 中包含 RETURNING 子句，此方式对任何 SQL 均有效：

client
  .query("DELETE FROM color RETURNING color_name")
  .execute()
  .onSuccess(rows -> {
    for (Row row : rows) {
      System.out.println("deleted color: " + row.getString("color_name"));
    }
  });

带有 RETURNING 子句的批量查询会生成一个 RowSet，其中为批处理中的每个元素包含一行：

client
  .preparedQuery("INSERT INTO color (color_name) VALUES ($1) RETURNING color_id")
  .executeBatch(Arrays.asList(Tuple.of("white"), Tuple.of("red"), Tuple.of("blue")))
  .onSuccess(res -> {
    for (RowSet<Row> rows = res; rows != null; rows = rows.next()) {
      Integer colorId = rows.iterator().next().getInteger("color_id");
      System.out.println("generated key: " + colorId);
    }
  });

使用独立连接

获取数据库连接

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

pool
  .getConnection()
  .compose(connection ->
    connection
      .preparedQuery("INSERT INTO Users (first_name,last_name) VALUES ($1, $2)")
      .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 $1")
  .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 ($1, $2)")
    .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 $1")
  .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 $1")
  .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）；已加载的行将缓存在内存中直到被消费，在此期间游标会暂停迭代抓取。

PostgreSQL 会在事务结束时销毁游标，因此游标 API 必须在事务内部使用，否则你很可能会收到 PostgreSQL 34000 错误。

查询链路追踪（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);

PostgreSQL 数据类型映射（PostgreSQL type mapping）

目前客户端支持以下 PostgreSQL 类型：

PostgreSQL 类型 Java 类型 Java 数组类型 说明

BOOLEAN

java.lang.Boolean

java.lang.Boolean[]

INT2

java.lang.Short

java.lang.Short[]

INT4

java.lang.Integer

java.lang.Integer[]

INT8

java.lang.Long

java.lang.Long[]

FLOAT4

java.lang.Float

java.lang.Float[]

FLOAT8

java.lang.Double

java.lang.Double[]

CHAR

java.lang.String

java.lang.Character[]

VARCHAR

java.lang.String

java.lang.String[]

TEXT

java.lang.String

java.lang.String[]

ENUM

java.lang.String

java.lang.String[]

NAME

java.lang.String

java.lang.String[]

SERIAL2

java.lang.Short

-

SERIAL4

java.lang.Integer

-

SERIAL8

java.lang.Long

-

NUMERIC

io.vertx.sqlclient.data.Numeric

io.vertx.sqlclient.data.Numeric[]

UUID

java.util.UUID

java.util.UUID[]

DATE

java.time.LocalDate

java.time.LocalDate[]

TIME

java.time.LocalTime

java.time.LocalTime[]

TIMETZ

java.time.OffsetTime

java.time.OffsetTime[]

TIMESTAMP

java.time.LocalDateTime

java.time.LocalDateTime[]

TIMESTAMPTZ

java.time.OffsetDateTime

java.time.OffsetDateTime[]

INTERVAL

io.vertx.pgclient.data.Interval

io.vertx.pgclient.data.Interval[]

BYTEA

io.vertx.core.buffer.Buffer

io.vertx.core.buffer.Buffer[]

JSON

java.lang.Object

java.lang.Object[]

参见 [handling-json]

JSONB

java.lang.Object

java.lang.Object[]

参见 [handling-json]

POINT

io.vertx.pgclient.data.Point

io.vertx.pgclient.data.Point[]

LINE

io.vertx.pgclient.data.Line

io.vertx.pgclient.data.Line[]

LSEG

io.vertx.pgclient.data.LineSegment

io.vertx.pgclient.data.LineSegment[]

BOX

io.vertx.pgclient.data.Box

io.vertx.pgclient.data.Box[]

PATH

io.vertx.pgclient.data.Path

io.vertx.pgclient.data.Path[]

POLYGON

io.vertx.pgclient.data.Polygon

io.vertx.pgclient.data.Polygon[]

CIRCLE

io.vertx.pgclient.data.Circle

io.vertx.pgclient.data.Circle[]

TSVECTOR

java.lang.String

java.lang.String[]

TSQUERY

java.lang.String

java.lang.String[]

INET

io.vertx.pgclient.data.Inet

io.vertx.pgclient.data.Inet[]

CIDR

io.vertx.pgclient.data.Cidr

io.vertx.pgclient.data.Cidr[]

MONEY

io.vertx.pgclient.data.Money

io.vertx.pgclient.data.Money[]

UNKNOWN

java.lang.String

java.lang.String[]

Tuple 解码在存储数值时会使用上述类型。在可能的情况下，它还会对实际值执行动态类型转换：

pool
  .query("SELECT 1::BIGINT "VAL"")
  .execute()
  .onComplete(ar -> {
    RowSet<Row> rowSet = ar.result();
    Row row = rowSet.iterator().next();

    // 存储为 java.lang.Long
    Object value = row.getValue(0);

    // 转换为 java.lang.Integer
    Integer intValue = row.getInteger(0);
  });

Tuple 编码在进行值编码时会采用上述类型映射，除非该类型为数值型，此时将改用 java.lang.Number：

pool
  .query("SELECT 1::BIGINT "VAL"")
  .execute()
  .onComplete(ar -> {
    RowSet<Row> rowSet = ar.result();
    Row row = rowSet.iterator().next();

    // 存储为 java.lang.Long
    Object value = row.getValue(0);

    // 转换为 java.lang.Integer
    Integer intValue = row.getInteger(0);
  });

处理 JSON

PostgreSQL JSON 和 JSONB 类型由以下 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); // 预期为 JsonObject

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

处理 NUMERIC

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

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

暂不支持的类型

以下类型当前暂不支持：

类型 说明

BIT

尚未实现

VARBIT

尚未实现

MACADDR

尚未实现

MACADDR8

尚未实现

XML

尚未实现

HSTORE

尚未实现

OID

尚未实现

VOID

尚未实现

处理数组（Handling arrays）

数组可在 Tuple 和 Row 上直接操作：

Tuple tuple = Tuple.of(new String[]{ "a", "tuple", "with", "arrays" });

// 向元组添加字符串数组
tuple.addArrayOfString(new String[]{"another", "array"});

// 获取第一个字符串数组
String[] array = tuple.getArrayOfStrings(0);

处理日期/时间无穷大（Handling Date/Time infinity）

PostgreSQL 定义了特殊值来表示正负无穷大（infinity）。

对应类型的 max/min 常量表示该 特殊值：

  • OffsetDateTime.MAX/OffsetDateTime.MIN

  • LocalDateTime.MAX/LocalDateTime.MIN

  • LocalDate.MAX/LocalDate.MIN

client
  .query("SELECT 'infinity'::DATE "LocalDate"")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      Row row = ar.result().iterator().next();
      System.out.println(row.getLocalDate("LocalDate").equals(LocalDate.MAX));
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

处理自定义类型（Handling custom types）

字符串用于表示自定义类型，无论是发送给 Postgres 还是从 Postgres 返回。

你可以从 PostgreSQL 中读取并以字符串形式获取自定义类型：

client
  .preparedQuery("SELECT address, (address).city FROM address_book WHERE id=$1")
  .execute(Tuple.of(3))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("Full Address " + row.getString(0) + ", City " + row.getString(1));
    }
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

你也可以通过提供字符串向 PostgreSQL 写入自定义类型：

client
  .preparedQuery("INSERT INTO address_book (id, address) VALUES ($1, $2)")
  .execute(Tuple.of(3, "('Anytown', 'Second Ave', false)"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    System.out.println(rows.rowCount());
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

处理全文搜索（Handling text search）

全文搜索使用 Java String 进行处理：

client
  .preparedQuery("SELECT to_tsvector( $1 ) @@ to_tsquery( $2 )")
  .execute(Tuple.of("fat cats ate fat rats", "fat & rat"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("Match : " + row.getBoolean(0));
    }
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

tsvector 和 tsquery 可以使用 Java String 从数据库中提取：

client
  .preparedQuery("SELECT to_tsvector( $1 ), to_tsquery( $2 )")
  .execute(Tuple.of("fat cats ate fat rats", "fat & rat"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("Vector : " + row.getString(0) + ", query : "+row.getString(1));
    }
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

处理枚举类型（Handling enumerated types）

PostgreSQL 枚举类型 会映射为 Java 字符串。

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

使用 Java 枚举类型（Using Java enum types）

你可以将 Java 枚举类型 映射到以下数据库列类型：

  • 字符串（VARCHAR、TEXT）

  • PostgreSQL 枚举类型

  • 数值（INT2、INT4、INT8）

client
  .preparedQuery("INSERT INTO colors VALUES ($1)")
  .execute(Tuple.of(Color.red))
  .flatMap(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"));
      }
    }
});

字符串和 PostgreSQL 枚举类型会与 Java 枚举通过 name() 方法返回的名称进行匹配。

数值类型会与 Java 枚举通过 ordinal() 方法返回的序号进行匹配。

使用收集器查询（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 对象的引用，因为在处理整个结果集时内部复用的是单一行实例。

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

发布 / 订阅（Pub/sub）

PostgreSQL 支持发布/订阅（Pub/sub）通信通道。

你可以设置 notificationHandler 来接收 PostgreSQL 的通知（notifications）：

connection.notificationHandler(notification -> {
  System.out.println("Received " + notification.getPayload() + " on channel " + notification.getChannel());
});

connection
  .query("LISTEN some-channel")
  .execute()
  .onComplete(ar -> {
  System.out.println("Subscribed to channel");
});

PgSubscriber 是一个通道管理器，它通过管理单个连接来提供按通道粒度的订阅功能：

PgSubscriber subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
);

// 你可以在连接前设置订阅通道
subscriber.channel("channel1").handler(payload -> {
  System.out.println("Received " + payload);
});

subscriber
  .connect()
  .onComplete(ar -> {
  if (ar.succeeded()) {

    // 或者在连接成功后设置订阅通道
    subscriber.channel("channel2").handler(payload -> {
      System.out.println("Received " + payload);
    });
  }
});

传递给 channel 方法的通道名称将作为 PostgreSQL 发送通知时使用的通道精确名称。请注意，这与 SQL 中的通道名称表示不同，在底层 PgSubscriber 会将提交的通道名预处理为带引号的标识符：

PgSubscriber subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
);

subscriber
  .connect()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // 复合通道名称 - 在 PostgreSQL 中需要用双引号标识符
      subscriber.channel("Complex.Channel.Name").handler(payload -> {
        System.out.println("Received " + payload);
      });
      subscriber.channel("Complex.Channel.Name").subscribeHandler(subscribed -> {
        subscriber.actualConnection()
          .query("NOTIFY "Complex.Channel.Name", 'msg'")
          .execute()
          .onComplete(notified -> {
            System.out.println("Notified "Complex.Channel.Name"");
          });
      });

      // PostgreSQL 简单标识符会被强制转换为小写
      subscriber.channel("simple_channel").handler(payload -> {
          System.out.println("Received " + payload);
      });
      subscriber.channel("simple_channel").subscribeHandler(subscribed -> {
        // 以下简单通道标识符会被强制转换为小写
        subscriber.actualConnection()
          .query("NOTIFY Simple_CHANNEL, 'msg'")
          .execute()
          .onComplete(notified -> {
            System.out.println("Notified simple_channel");
          });
      });

      // 以下通道名称超过了当前的限制
      // (NAMEDATALEN = 64) - 1 == 63 个字符，超出部分将被截断
      subscriber.channel("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbb")
        .handler(payload -> {
        System.out.println("Received " + payload);
      });
    }
  });

你可以提供一个重连策略函数，该函数接收重试次数 retries 作为参数，并返回重试等待时间 amountOfTime：

  • 当 amountOfTime < 0 时：订阅者关闭且不再进行重试

  • 当 amountOfTime = 0 时：订阅者立即尝试重新连接

  • 当 amountOfTime > 0 时：订阅者等待 amountOfTime 毫秒后尝试重新连接

PgSubscriber subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
);

// 最多重试 10 次，每次间隔 100 ms
subscriber.reconnectPolicy(retries -> {
  if (retries < 10) {
    return 100L;
  } else {
    return -1L;
  }
});

默认策略是不进行重连。

通知消息（Notice messages）

PostgreSQL 可以在连接生命周期内发送通知（notice）消息。

默认情况下，此类消息会在控制台中以 warn 级别记录。

你可以在连接上设置一个处理器来捕获这些消息并进行自定义处理：

connection.noticeHandler(notice -> {
  System.out.println("Received notice " + notice.getSeverity() + "" + notice.getMessage());
});

取消正在执行的请求（Cancelling Request）

PostgreSQL 支持取消正在执行中的请求。你可以使用 cancelRequest 取消传输中的请求。取消请求时会向服务器建立一个新连接并发出取消指令，随后关闭该新连接。

connection
  .query("SELECT pg_sleep(20)")
  .execute()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    // 假设这是一个长时间运行的查询并且仍在执行
    System.out.println("Query success");
  } else {
    // 服务器在收到取消请求后将中止当前查询
    System.out.println("Failed to query due to " + ar.cause().getMessage());
  }
});
connection
  .cancelRequest()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Cancelling request has been sent");
  } else {
    System.out.println("Failed to send cancelling request");
  }
});

取消信号可能会生效，也可能不会产生任何影响 —— 例如，如果信号在服务端已经处理完查询之后才到达，则不会产生任何效果。若取消成功，当前命令将提前终止并返回错误信息。

更多信息请参阅 PostgreSQL 官方文档。

使用 SSL/TLS

若要配置客户端使用 SSL 连接，你可以像配置 Vert.x NetClient 一样配置 PgConnectOptions。 客户端支持所有 SSL 模式，你可以通过 sslmode 进行配置。客户端默认处于 DISABLE SSL 模式。 ssl 参数仅作为设置 sslmode 的快捷方式：setSsl(true) 等同于 setSslMode(VERIFY_CA)，而 setSsl(false) 等同于 setSslMode(DISABLE)。

PgConnectOptions options = new PgConnectOptions()
  .setPort(5432)
  .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")));

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

SSL 协商（PostgreSQL 17+）

PostgreSQL 17 引入了直接 SSL 协商（Direct SSL negotiation），允许客户端在不进行传统协议握手的情况下立即建立 TLS 连接。 这减少了一个往返连接延迟，并允许标准 SSL 代理（nginx、haproxy）为 PostgreSQL 连接终结 TLS。

客户端通过 SslNegotiation 支持两种 SSL 协商模式：

  • POSTGRES（默认）：传统协商模式，客户端发送 SSL 请求并在建立 TLS 前等待服务端响应

  • DIRECT：直接协商模式，立即建立 TLS 连接（需要 PostgreSQL 17+）

PgConnectOptions options = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
  .setSslMode(SslMode.REQUIRE)
  .setSslNegotiation(SslNegotiation.DIRECT)
  .setSslOptions(new ClientSSLOptions()
    .setTrustOptions(new PemTrustOptions().addCertPath("/path/to/server.crt")));

PgConnection.connect(vertx, options)
  .onComplete(res -> {
    if (res.succeeded()) {
      System.out.println("Connected with direct SSL negotiation");
    } else {
      System.out.println("Could not connect: " + res.cause().getMessage());
    }
  });

你也可以通过连接 URI 配置 SSL 协商：

String connectionUri = "postgresql://localhost/mydb?sslmode=require&sslnegotiation=direct";
PgConnectOptions options = PgConnectOptions.fromUri(connectionUri);

或者使用环境变量：

export PGSSLNEGOTIATION=direct
直接 SSL 协商需要 PostgreSQL 17 或更高版本。在较旧的 PostgreSQL 版本中尝试使用 DIRECT 模式将导致 SSL 握手失败。

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

使用 4 层代理

你可以配置客户端使用 HTTP/1.x CONNECT、SOCKS4a 或 SOCKS5 4 层代理。

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

使用 7 层代理（Using a level 7 proxy）

7 层代理可以在连接到实际数据库的多个连接之间进行查询负载均衡。发生这种情况时，客户端可能会因为缺乏会话亲和性（session affinity）而产生异常，并引发类似 ERROR: unnamed prepared statement does not exist (26000) 等意外错误。

支持的代理：

  • 配置为 _pool_mode=transaction 的 PgBouncer。

你可以配置客户端以不同方式与代理进行交互：

connectOptions.setUseLayer7Proxy(true);

在此配置下，预编译语句无法被缓存，因此：

  • 必须禁用预编译语句缓存

  • 显式预编译语句只能在事务范围内有效，这意味着你可以使用游标，但游标所对应的预编译语句必须在事务范围内创建和销毁

高级连接池配置（Advanced pool configuration）