Cassandra java驱动程序 - 使用多线程提取数据时的高延迟

Posted 2023-03-22

【中文标题】Cassandra java驱动程序 - 使用多线程提取数据时的高延迟

【英文标题】：Cassandra java driver - high latency while extracting data with multiple threads

【发布时间】：2017-01-30 23:25:38

【问题描述】：

我可以看到 datastax cassandra 驱动程序 (3.0) 的奇怪行为。我创建了一个新集群，然后使用相同的集群对象启动了一组线程。如果我将线程保持在 1 或 2，我会看到 5 毫秒的平均提取时间，但如果我将线程增加到 60，则提取时间会增加到 200 毫秒（每个单线程）。奇怪的是，如果我让 60 个线程的应用程序运行并在同一台机器上启动另一个只有 1 个线程的进程，那么该单线程应用程序的提取时间又是 5 毫秒。所以这似乎与客户有关。我已经多次重复相同的测试以避免缓存冷启动问题。 以下是集群对象的配置方式：

    PoolingOptions poolingOptions = new PoolingOptions();
    poolingOptions
        .setConnectionsPerHost(HostDistance.LOCAL,  parallelism, parallelism+20)
        .setConnectionsPerHost(HostDistance.REMOTE, parallelism, parallelism+20)
        .setMaxRequestsPerConnection(HostDistance.LOCAL, 32768)
        .setMaxRequestsPerConnection(HostDistance.REMOTE, 2000);

    this.cluster = Cluster.builder()
            .addContactPoints(nodes)
            .withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)
            .withReconnectionPolicy(new ConstantReconnectionPolicy(100L))
            .withLoadBalancingPolicy(new TokenAwarePolicy(DCAwareRoundRobinPolicy.builder().build()))
            .withCompression(Compression.LZ4)
            .withPoolingOptions(poolingOptions)
            .withProtocolVersion(ProtocolVersion.V4)
            .build();

有人遇到过同样的问题吗？这似乎是一个客户端配置问题。可能还有一些 Netty 缺少的配置？

更新 1 应用程序正在使用以下查询提取数据块：

select * from table where id=? and ts>=? and ts<?

所以我有 60 个线程并行提取这些数据。 id 是分区键。每个查询都由线程执行为：

//Prepare statement
PreparedStatement stmt = ... get the prepared statment cached
BoundStatement bstmt = stmt.bind(...)
//Execute query
long te1 = System.nanoTime();       
ResultSet rs = this.session.execute(bstmt);
long te2 = System.nanoTime();
//Fetch...
Iterator<Row> iterator = rs.iterator();
while (!rs.isExhausted() && iterator.hasNext())  .... 

会话是一个并且跨所有线程共享。我正在测量的是 session.execute() 方法调用的平均时间。

谢谢！

更新 2 这是架构定义

CREATE TABLE d_t (
    id bigint,
    xid bigint,
    ts timestamp,
    avg double,
    ce double,
    cg double,
    p double,
    w double,
    c double,
    sum double,
    last double,
    max double,
    min double,
    p75 double,
    p90 double,
    p95 double,
    squad double,
    sumq double,
    wavg double,
    weight double,
    PRIMARY KEY ((id), xid, ts)
) WITH CLUSTERING ORDER BY (xid DESC, ts DESC)
and  compaction = 'class': 'SizeTieredCompactionStrategy' 
and gc_grace_seconds=86400
and caching =  'keys' : 'ALL', 'rows_per_partition':'36000' 
and min_index_interval = 2
and max_index_interval = 20;

更新 3 也尝试过

.setMaxRequestsPerConnection(HostDistance.LOCAL, 1)
.setMaxRequestsPerConnection(HostDistance.REMOTE, 1)

没有变化

【参考方案1】：

最终我认为这将取决于您的代码在做什么。你能分享一个例子吗？

关于延迟增加，您如何衡量？根据您的陈述：

奇怪的是，如果我让 60 个线程的应用程序运行并在同一台机器上启动另一个只有 1 个线程的进程，那么该单线程应用程序的提取时间又是 5 毫秒。

60 个并发请求确实不算多，一般来说，您不需要使用 datastax java 驱动程序为每个请求执行一个线程。您可以使用单个应用程序线程实现高吞吐量，因为驱动程序使用的 netty 事件循环组将完成大部分工作。

C* 使用的本机协议允许每个连接有多个请求。正如您在此处配置的那样，每个连接最多可以处理 32768 个并发请求。实际上，您根本不需要接触此配置，因为默认值（每个连接 1000 个请求）是合理的，因为实际上 C* 不会处理超过 cassandra.yaml 中的native_transport_max_threads（默认值 128）一次，然后将其余的排队。

因此，您不需要为每个主机建立很多连接。每个主机的默认 1 个核心连接对于 60 个并发请求应该绰绰有余。增加每台主机的连接数对您没有太大帮助，在分析中，我发现每台主机的连接数超过 8 个，吞吐量（数千个并发请求）和吞吐量在每台主机超过 16 个连接时变得更糟，尽管您的里程可能因环境而异。

话虽如此，我建议不要将PoolingOptions 配置为超出默认值，除了在您尝试实现更高吞吐量（> 10k 请求/秒）的情况下将 core 和 max 设置为 8 之外。

【讨论】：

谢谢！我已经用一个应用程序正在做什么的例子更新了这个问题。问候

在我开始之前，您的结果集有多大？ （每行的列数，每个查询有多少行？）

通常从 700 到 800 行，每行 15 列

还添加了架构定义

您是否需要更改以查看它？谢谢！