SpringBoot集成DynamoDB读写性能异常低下问题排查及优化咨询

嘿，刚看完你的问题，这种性能表现确实和DynamoDB应有的水平差太远了——正常情况下7k条数据完全应该在几秒到几十秒内搞定，咱们一步步拆解问题根源，给出优化方案：

一、核心问题：读写容量配置跟不上需求 #

你提到用了默认的1-10读写容量单位（RCU/WCU）+自动扩容，这是性能拉胯的头号原因：

• DynamoDB的1个WCU（写容量单位）每秒只能处理1条≤1KB的Item写入；1个RCU每秒只能读1条≤4KB的Item（强一致性）或者2条（最终一致性）。
• 自动扩容不是即时触发的：它需要连续几分钟检测到容量使用率超过阈值（默认70%）才会开始扩容，而且扩容速度有限制（比如初始阶段每次只能加少量容量）。
• 7k条数据按初始1WCU算，理论上要7000秒=116分钟，就算自动扩容到10WCU也要11分钟，再加上节流重试、网络开销，30分钟就不奇怪了；查询同理，RCU不够会导致多次重试、延迟飙升。

二、批量写入代码的关键优化点 #

你的写入代码有两个明显的问题：

1. 未处理未完成的写入项：batchWriteItem返回的unprocessedPutItems是因为节流导致写入失败的Item，你直接返回而没有重试，这会导致部分数据需要多次调用才能写入，拖慢整体速度。
2. 重复创建客户端资源：每次调用saveTransactions都重新创建DynamoDbTable和WriteBatch实例，这会带来不必要的初始化开销。

优化后的写入代码示例：

// 把DynamoDbTable实例缓存到类级别，不要每次方法调用都创建
private final DynamoDbTable<MyItems> dynamoDbTable;

// 通过构造函数注入客户端和表名
public YourRepository(DynamoDbEnhancedClient dynamoDbEnhancedClient, @Value("${dynamodb.table.name}") String tableName) {
    this.dynamoDbTable = dynamoDbEnhancedClient.table(tableName, TableSchema.fromBean(MyItems.class));
}

public List<MyItems> saveTransactions(List<MyItems> dynamoItems) {
    WriteBatch.Builder<MyItems> batchBuilder = WriteBatch.builder(MyItems.class).mappedTableResource(dynamoDbTable);
    dynamoItems.forEach(item -> batchBuilder.addPutItem(builder -> builder.item(item)));

    BatchWriteItemEnhancedRequest request = BatchWriteItemEnhancedRequest.builder()
            .writeBatches(batchBuilder.build())
            .build();

    List<MyItems> unprocessedItems = new ArrayList<>(dynamoItems);
    // 循环重试未处理的项，直到全部成功
    while (!unprocessedItems.isEmpty()) {
        BatchWriteResult result = dynamoDbEnhancedClient.batchWriteItem(request);
        unprocessedItems = result.unprocessedPutItemsForTable(dynamoDbTable);
        
        if (!unprocessedItems.isEmpty()) {
            // 重新构建包含未处理项的请求
            WriteBatch.Builder<MyItems> retryBatchBuilder = WriteBatch.builder(MyItems.class).mappedTableResource(dynamoDbTable);
            unprocessedItems.forEach(item -> retryBatchBuilder.addPutItem(builder -> builder.item(item)));
            request = BatchWriteItemEnhancedRequest.builder()
                    .writeBatches(retryBatchBuilder.build())
                    .build();
            
            // 加短暂延迟避免频繁节流
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("重试写入被中断", e);
            }
        }
    }
    return Collections.emptyList();
}

三、读取操作的优化建议 #

你的查询代码同样存在资源重复创建的问题，并且可能忽略了GSI的容量限制：

1. 缓存GSI实例：和主表一样，把DynamoDbIndex实例缓存到类级别，避免每次查询都重新初始化。
2. 检查GSI的容量配置：GSI有独立的读写容量配置，默认和主表一样是1-10，如果查询的QPS高，GSI的RCU不够会导致查询延迟飙升。
3. 优化查询分页：DynamoDB的查询是分页返回的，每次调用result.next()都会发起一次网络请求，如果数据量较大，可以考虑异步查询或者调整分页大小（通过pageSize()设置）减少请求次数。

优化后的查询代码示例：

private final DynamoDbIndex<MyItems> gsiIndex;

// 构造函数中初始化GSI
public YourRepository(DynamoDbEnhancedClient dynamoDbEnhancedClient, @Value("${dynamodb.table.name}") String tableName) {
    DynamoDbTable<MyItems> table = dynamoDbEnhancedClient.table(tableName, TableSchema.fromBean(MyItems.class));
    this.gsiIndex = table.index("GSI_INDEX");
}

public Optional<List<MyItems>> fetchTransactions(String itemCategory, String itemDate) {
    QueryConditional queryConditional = QueryConditional.keyEqualTo(
            Key.builder()
                    .partitionValue(itemCategory)
                    .sortValue(itemDate)
                    .build()
    );

    // 设置分页大小，减少网络请求次数
    Iterator<Page<MyItems>> result = gsiIndex.query(queryConditional)
            .pageSize(100) // 根据你的数据大小调整，最大1MB
            .iterator();

    List<MyItems> transactionsList = new ArrayList<>();
    while (result.hasNext()) {
        transactionsList.addAll(result.next().items());
    }
    return Optional.of(transactionsList);
}

四、其他潜在优化点 #

• 切换到异步客户端：使用DynamoDbAsyncEnhancedClient替代同步客户端，并行处理读写操作，能大幅提升吞吐量，尤其适合Kafka消费这种异步场景。
• 检查区域一致性：确保你的SpringBoot应用和DynamoDB表在同一个AWS区域，跨区域的网络延迟会严重拖慢性能。
• 评估Item大小：如果单个Item超过1KB，WCU/RCU的消耗会按比例增加（比如2KB的Item需要2个WCU），你可以通过AWS控制台查看Item大小分布，调整容量配置。
• 临时调高容量阈值：如果是一次性导入大量数据，可以临时把WCU的最小值调到100甚至更高，导入完成后再调回，避免自动扩容的滞后性。

最后验证点 #

你提到单批10条数据也要300ms，这个延迟已经包含了网络请求、客户端初始化、节流重试的开销，按照上面的优化（缓存实例、处理重试、调高容量），应该能降到几十ms以内。

备注：内容来源于stack exchange，提问作者springenthusiast