Cassandra 异步读写，最佳实践

Question

要设置上下文， 我们在 cassandra 中有 4 个表，其中 4 个是数据表，剩下的一个是搜索表（假设 DATA、SEARCH1、SEARCH2 和 SEARCH3 是表）。

我们有一个初始加载要求，一个请求中最多有 15k 行，用于 DATA 表，因此要保持搜索表同步。 我们以批量插入的方式执行此操作，每个 bacth 作为 4 个查询（每个表一个）以保持一致性。

但是对于每个批次，我们都需要读取数据。如果存在，只更新 DATA 表的 lastUpdatedDate 列，否则插入到所有 4 个表中。

下面是代码sn-p我们是怎么做的：

public List<Items> loadData(List<Items> items) {
    CountDownLatch latch = new CountDownLatch(items.size());
    ForkJoinPool pool = new ForkJoinPool(6);
    pool.submit(() -> items.parallelStream().forEach(item -> {
      BatchStatement batch = prepareBatchForCreateOrUpdate(item);
      batch.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
      ResultSetFuture future = getSession().executeAsync(batch);
      Futures.addCallback(future, new AsyncCallBack(latch), pool);
    }));

    try {
      latch.await();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }

    //TODO Consider what to do with the failed Items, Retry? or remove from the items in the return type
    return items;
}

private BatchStatement prepareBatchForCreateOrUpdate(Item item) {
    BatchStatement batch = new BatchStatement();
    Item existingItem = getExisting(item) //synchronous read
    if (null != data) {
      existingItem.setLastUpdatedDateTime(new Timestamp(System.currentTimeMillis()));
      batch.add(existingItem));
      return batch;
    }

    batch.add(item);
    batch.add(convertItemToSearch1(item));
    batch.add(convertItemToSearch2(item));
    batch.add(convertItemToSearch3(item));

    return batch;
  }

class AsyncCallBack implements FutureCallback<ResultSet> {
    private CountDownLatch latch;

    AsyncCallBack(CountDownLatch latch) {
      this.latch = latch;
    }

    // Cooldown the latch for either success or failure so that the thread that is waiting on latch.await() will know when all the asyncs are completed.
    @Override
    public void onSuccess(ResultSet result) {
      latch.countDown();
    }

    @Override
    public void onFailure(Throwable t) {
      LOGGER.warn("Failed async query execution, Cause:{}:{}", t.getCause(), t.getMessage());
      latch.countDown();
    }
  }

考虑到网络往返 b/w 应用程序和 cassandra 集群，15k 个项目的执行大约需要 1.5 到 2 分钟（两者都驻留在相同的 DNS 上，但 kubernetes 上的 pod 不同）

我们有想法使读取调用 getExisting(item) 也异步，但处理失败情况变得越来越复杂。 cassandra 的数据加载是否有更好的方法（仅考虑通过 datastax 企业 java 驱动程序的 Async wites）。

Answer 1

第一件事——Cassandra 中的批处理不同于关系数据库。通过使用它们，您会给集群带来更多负载。

关于使一切异步，我考虑了以下可能性：

1. 对数据库进行查询，获取Future 并向其添加侦听器 - 将在查询完成时执行（覆盖onSuccess）；
2. 通过该方法，您可以根据从 Cassandra 获得的结果安排下一个操作的执行。

您需要确保检查的一件事是，您不会同时发出过多的并发请求。在协议的第 3 版中，每个连接最多可以有 32k 个正在进行的请求，但在您的情况下，您最多可以发出 60k (4x15k) 个请求。我正在使用following wrapper around Session class 来限制进行中请求的数量。