数据库与持久化

你当前正在浏览历史版本 Vert.x 4.1.8 的文档。访问本页面的最新版本。

Vert.x Cassandra 客户端 (vertx-cassandra-client)

Vert.x Cassandra 客户端允许 Vert.x 应用程序以纯异步、非阻塞的方式直接与 Apache Cassandra 分布式 NoSQL 数据库集群进行通信交互。

快速上手

要在项目中使用该组件，请在构建描述文件中添加以下依赖：

Maven（在 pom.xml 中）：

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

Gradle（在 build.gradle 文件中）：

compile 'io.vertx:vertx-cassandra-client:5.0.12'

创建客户端实例

配置客户端选项

Cassandra 是一个典型的高可用分布式集群，通常拥有大量节点。要连接 Cassandra 集群，在构造 CassandraClientOptions 时需指定若干集群接触点（Contact Points）地址：

CassandraClientOptions options = new CassandraClientOptions()
  .addContactPoint("node1.address", 9142)
  .addContactPoint("node2.address", 9142)
  .addContactPoint("node3.address", 9142);
CassandraClient client = CassandraClient.create(vertx, options);

默认情况下，客户端会尝试连接本机的 9042 端口且不绑定特定 Keyspace。你可以通过如下选项进行定制：

CassandraClientOptions options = new CassandraClientOptions()
  .addContactPoint("localhost", 9142)
  .setKeyspace("my_keyspace");
CassandraClient client = CassandraClient.create(vertx, options);

若需底层驱动层面的深度调优，CassandraClientOptions 暴露了原生的 DataStax com.datastax.driver.core.Cluster.Builder 构建器。

共享客户端 (Shared Clients)

若应用中部署了多个 Verticle 实例，或有多个不同组件需要访问同一个 Cassandra 数据库，推荐创建共享客户端：

CassandraClientOptions options = new CassandraClientOptions()
  .addContactPoint("node1.address", 9142)
  .addContactPoint("node2.address", 9142)
  .addContactPoint("node3.address", 9142)
  .setKeyspace("my_keyspace");
CassandraClient client = CassandraClient.createShared(vertx, "sharedClientName", options);

具有相同名称的共享客户端将在底层复用同一个 com.datastax.driver.core.Session 会话连接。

客户端生命周期管理

客户端在创建后并不会立刻建立物理 TCP 连接，只有在首次执行查询操作时才会发起建连（懒加载）。

若同名的共享客户端已有其他实例执行过查询，新创建的共享客户端将即刻处于已连接状态。

在 Verticle 内部创建的客户端会在该 Verticle 被卸载（Undeployed）时由 Vert.x 自动停机销毁。换言之，你无须在 Verticle 的 stop 方法中手动调用 close。

在所有其他场景下，必须在应用退出时手动关闭客户端。

当某个共享客户端被关闭时，若系统中仍有其他同名共享客户端在运行，底层的 Driver Session 连接将保持打开状态不会被强制切断。

cassandraClient.queryStream("SELECT my_string_col FROM my_keyspace.my_table where my_key = 'my_value'")
  .onComplete(queryStream -> {
  if (queryStream.succeeded()) {
    CassandraRowStream stream = queryStream.result();

    // 当 HTTP 缓冲区排空可写时恢复数据流读取：
    response.drainHandler(v -> stream.resume());

    stream.handler(row -> {
      String value = row.getString("my_string_col");
      response.write(value);

      // 当 HTTP 写缓冲区已满时暂停 Cassandra 行读取（背压控制）：
      if (response.writeQueueFull()) {
        stream.pause();
      }
    });

    // 数据流读完后终止 HTTP 响应：
    stream.endHandler(end -> response.end());

  } else {
    queryStream.cause().printStackTrace();
    response
      .setStatusCode(500)
      .end("执行查询失败");
  }
});

2. 全量批量拉取 (Bulk Fetching)

当需要将全部查询结果一次性聚合加载到内存中时，可以使用全量拉取：

cassandraClient.executeWithFullFetch("SELECT * FROM my_keyspace.my_table where my_key = 'my_value'")
  .onComplete(executeWithFullFetch -> {
  if (executeWithFullFetch.succeeded()) {
    List<Row> rows = executeWithFullFetch.result();
    for (Row row : rows) {
      // 遍历处理每一行
    }
  } else {
    System.out.println("执行查询失败");
    executeWithFullFetch.cause().printStackTrace();
  }
});

仅在结果集较小或内存足以完整容纳整个结果列表时方可使用全量拉取。

结合 Java Stream Collector 收集器

支持使用标准 Java 收集器（Collector）直接转换查询结果：

cassandraClient.execute("SELECT * FROM users", listCollector)
  .onComplete(ar -> {
    if (ar.succeeded()) {
      String list = ar.result();
      System.out.println("收集结果: " + list);
    } else {
      System.out.println("失败: " + ar.cause().getMessage());
    }
  });

3. 底层分页拉取 (Low Level Fetch)

提供对分页拉取（Pagination）的底层细粒度掌控：

cassandraClient.execute("SELECT * FROM my_keyspace.my_table where my_key = 'my_value'")
  .onComplete(execute -> {
  if (execute.succeeded()) {
    ResultSet resultSet = execute.result();

    if (resultSet.remaining() != 0) {
      Row row = resultSet.one();
      System.out.println("成功读取单行数据");
    } else if (!resultSet.hasMorePages()) {
      System.out.println("已无后续分页数据");
    } else {
      resultSet.fetchNextPage().onComplete(fetchMoreResults -> {
        if (fetchMoreResults.succeeded()) {
          int availableWithoutFetching = resultSet.remaining();
          System.out.println("当前已拉取但尚未消费的行数: " + availableWithoutFetching);
        } else {
          fetchMoreResults.cause().printStackTrace();
        }
      });
    }
  } else {
    execute.cause().printStackTrace();
  }
});

预编译语句 (Prepared Queries)

出于防止 SQL 注入风险与提升执行效率，对复用频率高的查询强烈建议使用预编译语句：

cassandraClient.prepare("SELECT * FROM my_keyspace.my_table where my_key = ? ")
  .onComplete(preparedStatementResult -> {
    if (preparedStatementResult.succeeded()) {
      PreparedStatement preparedStatement = preparedStatementResult.result();
      // 后续可直接绑定参数执行：
      cassandraClient.execute(preparedStatement.bind("my_value"))
        .onComplete(done -> {
          ResultSet results = done.result();
        });
    }
  });

批量操作 (Batching)

若需一次性原子提交多条写操作，可以使用 BatchStatement：

BatchStatement batchStatement = BatchStatement.newInstance(BatchType.LOGGED)
  .add(SimpleStatement.newInstance("INSERT INTO NAMES (name) VALUES ('Pavel')"))
  .add(SimpleStatement.newInstance("INSERT INTO NAMES (name) VALUES ('Thomas')"))
  .add(SimpleStatement.newInstance("INSERT INTO NAMES (name) VALUES ('Julien')"));

cassandraClient
  .execute(batchStatement)
  .onComplete(result -> {
    if (result.succeeded()) {
      System.out.println("Batch 批处理执行成功！");
    } else {
      result.cause().printStackTrace();
    }
  });

链路追踪支持 (Tracing)

当全局启用了 Vert.x Tracing 时，Cassandra 客户端会自动为查询操作上报 Span：

默认追踪策略为 PROPAGATE，可通过 CassandraClientOptions.setTracingPolicy(TracingPolicy.ALWAYS) 强制始终上报。

Redis

JDBC

Vert.x Cassandra 客户端 (vertx-cassandra-client)

快速上手

创建客户端实例

配置客户端选项

共享客户端 (Shared Clients)

客户端生命周期管理

使用查询 API

查询操作

1. 响应式流式读取 (Streaming)

2. 全量批量拉取 (Bulk Fetching)

结合 Java Stream Collector 收集器

3. 底层分页拉取 (Low Level Fetch)

预编译语句 (Prepared Queries)

批量操作 (Batching)

链路追踪支持 (Tracing)