（Datastax 4.1.0）（Cassandra）如何使用 session.executeAsync 收集所有响应？答案

【问题标题】：(Datastax 4.1.0) (Cassandra )How do I collect all responses with session.executeAsync?（Datastax 4.1.0）（Cassandra）如何使用 session.executeAsync 收集所有响应？
【发布时间】：2020-05-24 08:53:35
【问题描述】：

我想通过 execute.Async 调用对 cassandra db 进行异步调用在 manuel 中，我找到了这段代码，但我不明白如何将所有行收集到任何列表中。非常基本的调用，例如从表中选择 *，我想存储所有结果。

https://docs.datastax.com/en/developer/java-driver/4.4/manual/core/async/

CompletionStage<CqlSession> sessionStage = CqlSession.builder().buildAsync();

// Chain one async operation after another:
CompletionStage<AsyncResultSet> responseStage =
    sessionStage.thenCompose(
        session -> session.executeAsync("SELECT release_version FROM system.local"));

// Apply a synchronous computation:
CompletionStage<String> resultStage =
    responseStage.thenApply(resultSet -> resultSet.one().getString("release_version"));

// Perform an action once a stage is complete:
resultStage.whenComplete(
    (version, error) -> {
      if (error != null) {
        System.out.printf("Failed to retrieve the version: %s%n", error.getMessage());
      } else {
        System.out.printf("Server version: %s%n", version);
      }
      sessionStage.thenAccept(CqlSession::closeAsync);
    });

【问题讨论】：

标签： java cassandra completable-future datastax-java-driver completion-stage

【解决方案1】：

您需要参考section about asynchronous paging - 您需要提供一个回调，将数据收集到作为外部对象提供的列表中。文档有以下示例：

CompletionStage<AsyncResultSet> futureRs =
    session.executeAsync("SELECT * FROM myTable WHERE id = 1");
futureRs.whenComplete(this::processRows);

void processRows(AsyncResultSet rs, Throwable error) {
  if (error != null) {
    // The query failed, process the error
  } else {
    for (Row row : rs.currentPage()) {
      // Process the row...
    }
    if (rs.hasMorePages()) {
      rs.fetchNextPage().whenComplete(this::processRows);
    }
  }
}

在这种情况下，processRows 可以将数据存储在作为当前对象一部分的列表中，如下所示：

class Abc {
  List<Row> rows = new ArrayList<>();

  // call to executeAsync

  void processRows(AsyncResultSet rs, Throwable error) {
....
    for (Row row : rs.currentPage()) {
      rows.add(row);
    }
....

  }
}

但是您需要非常小心select * from table，因为它可能会返回很多结果，而且如果您有太多数据，它可能会超时 - 在这种情况下，最好执行令牌范围扫描（我有 @ 987654322@，但 4.x 还没有）。

【讨论】：

我们如何通过分页按集群键的顺序收集结果？是否保证井井有条
Cassandra 保证数据会按集群键排序，但只在分区内...
我明白了.. 同时对此stackoverflow.com/questions/63598069/… 的任何想法我正在寻找简单的 Row 到 POJO 映射器，所以我不需要输入代码。它在 datastax 驱动程序中是否可用？

【解决方案2】：

这是 4.x 的示例（您还可以找到 4.4 BTW 提供的响应式代码示例）

https://github.com/datastax/cassandra-reactive-demo/blob/master/2_async/src/main/java/com/datastax/demo/async/repository/AsyncStockRepository.java

【讨论】：