如何修改Gatling代码以实现每秒执行尽可能多的CQL请求？

原测试代码 #

val feeder = Iterator.continually(
    Map(
      "id" -> serialGen(),
      "randomString" -> random.nextString(20),
      "randomNum" -> jrandom.nextLong(),
      "randomMap" -> maps(jrandom.nextInt(maps.size))
    ))

val scn = scenario("Two statements").repeat(1) { 
    feed(feeder)
      .exec(cql("simple SELECT")
        .execute("SELECT * FROM test_table WHERE id = ${randomNum}")
        .consistencyLevel(ConsistencyLevel.LOCAL_QUORUM)
        .check(rowCount.satisfies(a => a >= 0)))
      .exec(cql("prepared INSERT")
        .execute(prepared)
        .withParams("${id}", "${randomMap}", "${randomString}")
        .consistencyLevel(ConsistencyLevel.LOCAL_QUORUM))
}

setUp(scn.inject(
    rampUsersPerSec(1) to 1 during (1.minute)
)).protocols(cqlConfig)

当前repeat(1)时每个用户每秒仅执行2条CQL语句，若要最大化每秒执行的CQL语句数量，可从以下几个维度修改：

1. 大幅提升并发用户数 #

原注入策略仅维持1个并发用户，这是吞吐量的核心瓶颈。直接调整为高并发的用户注入模式，比如：

setUp(scn.inject(
    // 快速扩容到目标并发数后持续压测
    rampUsersPerSec(10) to 500 during (2.minutes),
    // 或者直接维持固定高并发
    constantUsersPerSec(500) during (10.minutes)
)).protocols(cqlConfig)

注：目标并发数需根据Cassandra集群的承载能力调整，避免压垮集群。

2. 并行执行CQL请求 #

原代码中SELECT和INSERT是串行执行的，改为并行执行可提升单用户的吞吐量：

val scn = scenario("Two statements").repeat(300) { 
    feed(feeder)
      .exec(
        forkJoin(
          cql("simple SELECT")
            .execute("SELECT * FROM test_table WHERE id = ${randomNum}")
            .consistencyLevel(ConsistencyLevel.LOCAL_QUORUM),
          cql("prepared INSERT")
            .execute(prepared)
            .withParams("${id}", "${randomMap}", "${randomString}")
            .consistencyLevel(ConsistencyLevel.LOCAL_QUORUM)
        )
      )
}

3. 移除不必要的性能开销 #

• 去掉SELECT请求中的check(rowCount.satisfies(a => a >= 0))：这类检查会增加客户端CPU消耗，若仅关注吞吐量，无需验证返回结果。
• 调整Feeder位置：若不需要每次循环都生成全新数据，可将feed(feeder)移到repeat循环外，减少重复的数据生成开销：

val scn = scenario("Two statements")
    .feed(feeder)
    .repeat(300) { 
      .exec(...) // 并行执行CQL请求
    }

4. 优化CQL客户端配置 #

调整Cassandra连接池参数，提升并发连接能力：

val cqlConfig = cql.session
    .withClusterConfig(cluster => cluster
      .withPoolingOptions(new PoolingOptions()
        .setCoreConnectionsPerHost(HostDistance.LOCAL, 10)
        .setMaxConnectionsPerHost(HostDistance.LOCAL, 20)
        .setMaxRequestsPerConnection(HostDistance.LOCAL, 1000))
    .build()

5. 优化JVM参数 #

为Gatling进程配置足够的内存和高效的GC策略，避免GC停顿影响测试：

# 示例启动参数
java -Xms8G -Xmx8G -XX:+UseG1GC -jar gatling-charts-highcharts-bundle-xxx.jar

优化后示例代码 #

val feeder = Iterator.continually(
    Map(
      "id" -> serialGen(),
      "randomString" -> random.nextString(20),
      "randomNum" -> jrandom.nextLong(),
      "randomMap" -> maps(jrandom.nextInt(maps.size))
    ))

// 预编译SELECT语句提升性能
val selectPrepared = session.prepare("SELECT * FROM test_table WHERE id = ?")

val scn = scenario("Parallel CQL Statements").repeat(300) { 
    feed(feeder)
      .exec(
        forkJoin(
          cql("prepared SELECT")
            .execute(selectPrepared)
            .withParams("${randomNum}")
            .consistencyLevel(ConsistencyLevel.LOCAL_QUORUM),
          cql("prepared INSERT")
            .execute(prepared)
            .withParams("${id}", "${randomMap}", "${randomString}")
            .consistencyLevel(ConsistencyLevel.LOCAL_QUORUM)
        )
      )
}

val cqlConfig = cql.session
    .withClusterConfig(cluster => cluster
      .withPoolingOptions(new PoolingOptions()
        .setCoreConnectionsPerHost(HostDistance.LOCAL, 10)
        .setMaxConnectionsPerHost(HostDistance.LOCAL, 20)
        .setMaxRequestsPerConnection(HostDistance.LOCAL, 1000))
    .build()

setUp(scn.inject(
    rampUsersPerSec(10) to 500 during (2.minutes),
    constantUsersPerSec(500) during (10.minutes)
)).protocols(cqlConfig)

内容的提问来源于stack exchange，提问作者Pavel Orekhov