如何通过Java Executor框架优化本地DynamoDB批量插入速率？

针对你遇到的5500条数据写入耗时超100秒的问题，结合本地DynamoDB的特性和你的代码，我整理了几个关键优化点：

1. 先搞懂本地DynamoDB的吞吐量特性 #

本地DynamoDB默认用SQLite做存储后端，你设置的ProvisionedThroughput（读写容量单位）是不生效的——官方明确说明本地模式下不会强制执行吞吐量限制，所以调整WriteCapacityUnit没变化是正常的，别在这白费功夫啦。

2. 修复代码里的批次遗漏漏洞 #

你的代码在循环结束后，没处理剩余的不足25条的批次（这次5500刚好是25的倍数，但逻辑上有问题），赶紧补上这段：

// 循环结束后处理剩余的未提交批次
if (!batchList.isEmpty()) {
    Map<String, List<WriteRequest>> batchTableRequests = new HashMap<>();
    batchTableRequests.put("Employee", batchList);
    BatchWriteItemRequest batchWriteItemRequest = new BatchWriteItemRequest();
    batchWriteItemRequest.setRequestItems(batchTableRequests);
    listOfBatchItemsRequest.add(batchWriteItemRequest);
}

3. 必须处理未完成的写入请求 #

batchWriteItem调用可能会返回未处理的请求（本地模式下SQLite的锁机制可能导致部分请求被阻塞），你现在的代码直接忽略了这些，不仅效率低还可能丢数据。修改Callable任务，加上重试逻辑：

listOfBatchItemsRequest.stream().map(batchItemsRequest -> service.submit(() -> {
    BatchWriteItemResult result;
    Map<String, List<WriteRequest>> unprocessedItems = batchItemsRequest.getRequestItems();
    do {
        result = aws.batchWriteItem(new BatchWriteItemRequest().withRequestItems(unprocessedItems));
        unprocessedItems = result.getUnprocessedItems();
        // 加个短暂延迟，避免频繁重试浪费资源
        if (!unprocessedItems.isEmpty()) {
            Thread.sleep(100);
        }
    } while (!unprocessedItems.isEmpty());
    return result;
})).collect(Collectors.toList());

4. 别给本地DynamoDB开太多线程 #

本地DynamoDB基于SQLite，而SQLite是单线程写的（写操作会独占锁），开20个线程只会导致大量锁等待和上下文切换，反而拖慢速度。建议把线程池大小调到4-8试试：

ExecutorService service = Executors.newFixedThreadPool(4);

5. 用内存模式跑本地DynamoDB #

默认本地DynamoDB会把数据写到磁盘，启动时加-inMemory参数，让数据存在内存里，写入速度会飞起来：

java -Djava.library.path=./DynamoDBLocal_lib -jar DynamoDBLocal.jar -inMemory

6. 简化客户端初始化，别搞冗余 #

你代码里创建了两个AmazonDynamoDBClient实例，其中一个完全没用，删了它，用新版的构建器更稳定：

AmazonDynamoDB aws = AmazonDynamoDBClientBuilder.standard()
        .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration("http://localhost:8000", "us-east-1"))
        .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("x", "y")))
        .build();

7. 优化数据转换逻辑 #

你用ItemUtils.fromSimpleMap转数据，会用到反射，开销不小。如果你的JSON结构固定，直接手动构造Item更快：

JSONObject emp = iterator.next();
PutRequest putRequest = new PutRequest()
        .withItem(new HashMap<String, AttributeValue>() {{
            put("ID", new AttributeValue().withS(emp.getString("ID")));
            put("Name", new AttributeValue().withS(emp.getString("Name")));
            // 其他字段按实际情况补充
        }});

最后测试效果 #

按上面的优化点调整后，内存模式下5500条数据的写入时间应该能控制在10秒以内，试试吧！

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