Amazon Keyspaces高写入遇PT2S超时，如何达CloudSearch写入速率？

针对当前写入速率低、移除休眠后触发超时的问题，结合Keyspaces的特性，可通过以下方向优化：

1. 移除固定休眠，改用并发请求限流 #

代码中的Thread.sleep(100)直接限制了写入速率，移除后无限制发起异步请求会导致驱动过载触发超时。可使用Semaphore控制并发请求数，避免一次性压垮服务或客户端：

// 初始化信号量，并发数可根据测试调整，建议从200开始尝试
private final Semaphore writeSemaphore = new Semaphore(200);

public void run() {
    while (true) {
        JsonNode record = null;
        try {
            record = queue.take();
            writeSemaphore.acquire(); // 获取许可，无许可则阻塞
            awsKeyspacesService.uploadRecord(record, tableName, writeSemaphore);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

// 修改上传方法，完成后释放信号量
public void uploadRecord(JsonNode record, String table, Semaphore semaphore) {
    InsertInto insertInto = insertInto(keyspacesWriterConfig.getKeyspaceName(), table);
    SimpleStatement simpleStatement = insertInto
            .json(record.toString())
            .build()
            .setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);

    CompletionStage<AsyncResultSet> completionStage = session.executeAsync(simpleStatement);
    completionStage.thenAccept(result -> {
        logger.info("Wrote {} to {}", record.get("ID"), table);
        semaphore.release(); // 完成后释放许可
    });
    completionStage.exceptionally(throwable -> {
        logger.warn("Error {} to {}: {}", record.get("ID"), table, throwable.getMessage());
        semaphore.release(); // 异常时也要释放许可，避免死锁
        return null;
    });
}

2. 优化Datastax驱动配置 #

调整驱动参数适配Keyspaces特性：

• 增大连接池大小：设置advanced.connection.pool.size为10-20（根据实例规格调整），确保有足够连接处理并发请求
• 合理调整超时：将request.timeout从默认2秒调整为5秒，避免正常请求因网络波动触发超时
• 启用重试策略：配置advanced.retry-policy为DefaultRetryPolicy.INSTANCE，允许重试幂等的写入请求

3. 启用多线程消费队列 #

单线程消费无法充分利用Keyspaces的吞吐量，可启动4-8个消费线程并行处理队列中的记录：

// 初始化多个消费线程
for (int i = 0; i < 4; i++) {
    new Thread(yourRunnableInstance).start();
}

4. 确认Keyspaces表的吞吐量配置 #

检查表的**写入吞吐量（WRU）**设置：

• 你每行数据322字节，每条写入消耗1 WRU，目标500条/秒需至少配置500 WRU
• 若当前配置不足，需在AWS控制台调整表的吞吐量模式（按需或预置），确保WRU能支撑目标速率

5. 优化数据写入方式 #

• 替换JSON插入：相比CQL JSON插入，使用PreparedStatement绑定参数的方式性能更高，减少序列化开销：

// 全局预编译语句（初始化一次即可）
private final PreparedStatement insertStmt = session.prepare(
    "INSERT INTO keyspace.table (id, col1, col2) VALUES (?, ?, ?)"
);

// 写入时绑定参数
public void uploadRecord(JsonNode record, String table, Semaphore semaphore) {
    BoundStatement boundStmt = insertStmt.bind(
        record.get("ID").asText(),
        record.get("col1").asText(),
        record.get("col2").asInt()
    ).setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);

    CompletionStage<AsyncResultSet> completionStage = session.executeAsync(boundStmt);
    // 后续处理同之前的信号量释放逻辑...
}

• 避免跨分区Batch：跨分区Batch会降低性能，保持单条异步写入+并发控制是更优选择

6. 监控与调优 #

• 通过AWS CloudWatch监控Keyspaces的WriteThrottleEvents指标，若出现限流，说明WRU配置不足，需继续提升
• 监控客户端驱动的requests-per-second和timeout-rate指标，动态调整Semaphore并发数和驱动配置

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