如何使用异步查询提高 cassandra 3.0 读取性能和吞吐量？

Question

我有一张桌子：

CREATE TABLE my_table (
    user_id text,
    ad_id text,
    date timestamp,
    PRIMARY KEY (user_id, ad_id)
);

我使用的 user_id 和 ad_id 的长度不超过 15 个字符。

我这样查询表：

Set<String> users = ... filled somewhere 
Session session = ... builded somewhere
BoundStatement boundQuery = ... builded somewhere
(using query: "SELECT * FROM my_table WHERE user_id=?")

List<Row> rowAds = 
      users.stream()
          .map(user -> session.executeAsync(boundQuery.bind(user)))
          .map(ResultSetFuture::getUninterruptibly)
          .map(ResultSet::all)
          .flatMap(List::stream)
          .collect(toList());

用户集大约有 3000 个元素，每个用户大约有 300 个广告。

此代码在同一台机器的 50 个线程中执行，（使用不同的用户），（使用相同的 Session 对象）

算法需要 2 到 3 秒才能完成

Cassandra 集群有 3 个节点，复制因子为 2。每个节点有 6 个核心和 12 GB 内存。

Cassandra 节点的 CPU 容量为 60%，内存为 33%，内存为 66%（包括页面缓存）
查询机器是它的cpu容量的50%，内存的50%

如何将读取时间缩短到 1 秒以下？

谢谢！

更新：

在回答了一些问题后（非常感谢），我意识到我没有并行执行查询，所以我将代码更改为：

List<Row> rowAds = 
     users.stream()
       .map(user ->  session.executeAsync(boundQuery.bind(user)))
       .collect(toList())
       .stream()
       .map(ResultSetFuture::getUninterruptibly)
       .map(ResultSet::all)
       .flatMap(List::stream)
       .collect(toList());

所以现在查询是并行完成的，这给了我大约 300 毫秒的时间，这让我有了很大的进步！
但我的问题还在继续，它可以更快吗？ 再次感谢！

Answer 1

users.stream()
          .map(user -> session.executeAsync(boundQuery.bind(user)))
          .map(ResultSetFuture::getUninterruptibly)
          .map(ResultSet::all)
          .flatMap(List::stream)
          .collect(toList());

一句话。在第二个map() 你打电话给ResultSetFuture::getUninterruptibly。这是一个阻塞调用，因此您不会从异步 exec 中受益多少......

相反，尝试将驱动程序返回的 Futures 列表（提示：ResultSetFuture 正在实现 Guava 的 ListenableFuture 接口）转换为 List 的 Future

见：http://docs.guava-libraries.googlecode.com/git/javadoc/com/google/common/util/concurrent/Futures.html#successfulAsList(java.lang.Iterable)