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

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

Reactive Oracle Client 是一款专为 Oracle 数据库打造的客户端，提供了简洁易用的 API，专注于高伸缩性与低开销。

特性（Features）

  • 事件驱动

  • 内置连接池

  • 支持 Java 8 日期和时间（Date and Time）

  • 支持 SSL/TLS

  • 提供 RxJava API

  • 游标（Cursor）

  • 行流式处理（Row streaming）

暂不支持的功能（Not supported yet）

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

  • 存储过程（Stored Procedures）

使用方法

要使用 Reactive Oracle Client，请将以下依赖项添加至项目构建描述符的 dependencies 部分：

  • Maven（在 pom.xml 中）：

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-oracle-client</artifactId>
  <version>5.1.10</version>
</dependency>

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

dependencies {
  compile 'io.vertx:vertx-oracle-client:5.1.10'
}

快速入门

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

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

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

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

连接至 Oracle 数据库

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

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

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

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

基于连接池的客户端使用连接池来管理连接，任何操作都将从连接池中借用一个连接来执行，并在执行完毕后将其归还给连接池。

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

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

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

当你不再需要该连接池时，需要将其释放关闭：

pool.close();

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

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

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

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

// 创建基于连接池的客户端
Pool client = OracleBuilder.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());
  }
});

连接使用完毕后，你必须将其关闭以将其释放并归还给连接池，以便复用。

连接池共享

你可以在多个 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 = OracleBuilder.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 = OracleBuilder.pool()
  .with(new PoolOptions()
    .setMaxSize(maxSize)
    .setShared(true)
    .setName("my-pool")
    .setEventLoopSize(4))
  .connectingTo(database)
  .using(vertx)
  .build();

配置

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

数据对象（Data Object）

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

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

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

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

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

连接 URI

除了使用 OracleConnectOptions 数据对象进行配置之外，我们还提供了一种通过连接 URI 进行配置的替代方式。

使用 EZConnect 格式：

String connectionUri = "oracle:thin:@mydbhost1:5521/mydbservice?connect_timeout=10sec";

// 连接选项
OracleConnectOptions connectOptions = OracleConnectOptions.fromUri(connectionUri)
  .setUser("user")
  .setPassword("secret");

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

// 基于连接 URI 创建连接池
Pool pool = OracleBuilder.pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

或者使用 TNS 别名（TNS Alias）格式：

String connectionUri = "oracle:thin:@prod_db?TNS_ADMIN=/work/tns/";

// 连接选项
OracleConnectOptions connectOptions = OracleConnectOptions.fromUri(connectionUri)
  .setUser("user")
  .setPassword("secret");

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

// 基于连接 URI 创建连接池
Pool pool = OracleBuilder.pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

连接重试

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

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

获取生成的主键值

在执行 INSERT 查询时，你可以获取生成的主键值。

这些值作为 Row 实例返回。 该实例可以通过使用 OracleClient.GENERATED_KEYS 属性类型调用 SqlResult.property(kind) 来获取。

可以按列名检索生成的主键值：

String sql = "INSERT INTO EntityWithIdentity (name, position) VALUES (?, ?)";

// 按列名检索生成的键列值
OraclePrepareOptions options = new OraclePrepareOptions()
  .setAutoGeneratedKeysIndexes(new JsonArray().add("ID"));

client.preparedQuery(sql, options)
  .execute(Tuple.of("john", 3))
  .onComplete(ar -> {
    if (ar.succeeded()) {
      RowSet<Row> result = ar.result();

      Row generated = result.property(OracleClient.GENERATED_KEYS);
      Long id = generated.getLong("ID");
    }
  });

或者，也可以按列索引进行检索：

String sql = "INSERT INTO EntityWithIdentity (name, position) VALUES (?, ?)";

// 按列索引检索生成的键列值
OraclePrepareOptions options = new OraclePrepareOptions()
  .setAutoGeneratedKeysIndexes(new JsonArray().add("1"));

client.preparedQuery(sql, options)
  .execute(Tuple.of("john", 3))
  .onComplete(ar -> {
    if (ar.succeeded()) {
      RowSet<Row> result = ar.result();

      Row generated = result.property(OracleClient.GENERATED_KEYS);
      Long id = generated.getLong("ID");
    }
  });

使用独立连接

获取数据库连接

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

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 = ps.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）；已加载的行将缓存在内存中直到被消费，在此期间游标会暂停迭代抓取。

Oracle 数据类型映射

目前客户端支持以下 Oracle 数据类型：

Oracle 类型 Java 类型 说明

CHAR / VARCHAR2

java.lang.String

NCHAR / NVARCHAR2

java.lang.String

Unicode 字符串

NUMBER

java.math.BigDecimal

FLOAT

java.lang.Double

DATE

java.time.LocalDate

TIMESTAMP

java.time.LocalDateTime

RAW

io.vertx.core.buffer.Buffer

BLOB

io.vertx.core.buffer.Buffer

参见 处理 BLOB（Handling BLOB）

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

处理 BLOB（Handling BLOB）

虽然支持 BLOB 数据类型，但有一点需要注意：在写入或过滤时，必须由 Blob 实例表示。 然而，在读取 BLOB 数据时，客户端返回的是 Buffer。

client.preparedQuery("INSERT INTO images (name, data) VALUES (?, ?)")
  // 插入数据时使用 io.vertx.oracleclient.data.Blob
  .execute(Tuple.of("beautiful-sunset.jpg", Blob.copy(imageBuffer)))
  .onComplete(ar -> {
    // 执行相应逻辑
  });

client.preparedQuery("SELECT data FROM images WHERE id = ?")
  .execute(Tuple.of(id))
  .onComplete(ar -> {
    if (ar.succeeded()) {
      Row row = ar.result().iterator().next();

      // 读取数据时使用 io.vertx.core.buffer.Buffer
      Buffer data = row.getBuffer("data");
    }
  });

查询链路追踪

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

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

  • 操作名称为 QueryMessage

  • 包含以下 Tags：

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

  • db.user：数据库用户名

  • db.instance：数据库实例

  • db.statement：SQL 查询语句

  • db.type：固定为 sql

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

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

options.setTracingPolicy(TracingPolicy.ALWAYS);

使用收集器（Collector）执行查询

你可以在查询 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());
    }
  });

使用 SSL/TLS

若要在客户端选项中启用加密，请使用 setSsl 方法。 默认情况下，ssl 为 false。

oracleConnectOptions.setSsl(true);

可以通过属性自定义加密配置。 例如，设置信任库（Trust store）：

oracleConnectOptions
  .setSsl(true)
  .addProperty("javax.net.ssl.trustStore", pathToTrustStore)
  .addProperty("javax.net.ssl.trustStoreType", "JKS")
  .addProperty("javax.net.ssl.trustStorePassword", trustStorePassword);