Java AWS Lambda异步写入Kinesis Stream的完整性保障问题

这个问题我之前做Kinesis集成时也踩过坑——Lambda的执行模型就是主线程跑完就立刻终止运行环境，后台的异步任务自然会被强制打断，最后几个请求没写完太正常了。要解决这个问题，核心就是让Lambda的主线程等待所有异步Kinesis写入任务完成后再结束，下面给你几个实用的实现方式：

方案1：收集Future对象逐个等待（AWS SDK v1适用） #

putRecordAsync()会返回一个Future<PutRecordResult>，我们可以把所有请求的Future都存起来，然后挨个调用get()阻塞等待结果：

// 初始化Kinesis异步客户端
AmazonKinesisAsync kinesisClient = AmazonKinesisAsyncClientBuilder.defaultClient();
List<Future<PutRecordResult>> writeFutures = new ArrayList<>();

// 批量提交异步写入请求
for (YourRecord dataRecord : recordsToSend) {
    PutRecordRequest request = new PutRecordRequest()
        .withStreamName("your-target-stream")
        .withData(ByteBuffer.wrap(dataRecord.getContent()))
        .withPartitionKey(dataRecord.getPartitionKey());
    
    Future<PutRecordResult> future = kinesisClient.putRecordAsync(request);
    writeFutures.add(future);
}

// 等待所有异步任务完成
for (Future<PutRecordResult> future : writeFutures) {
    try {
        // get()会阻塞直到任务完成或抛出异常
        PutRecordResult result = future.get();
        // 可选：记录成功写入的序列号
        System.out.println("写入成功，SequenceNumber: " + result.getSequenceNumber());
    } catch (InterruptedException | ExecutionException e) {
        // 处理写入失败的情况，比如打日志、重试或抛出异常终止Lambda
        System.err.println("Kinesis写入失败: " + e.getMessage());
        // 如果业务要求不能丢数据，这里可以抛出异常触发Lambda重试
        // throw new RuntimeException("Kinesis批量写入失败", e);
    }
}

这种方式最直接，适合用SDK v1的场景，注意要处理异常，避免单个请求失败导致整个流程卡住（或者根据业务需求决定是否终止）。

方案2：用CompletableFuture.allOf()批量等待（AWS SDK v2推荐） #

如果你已经升级到AWS SDK v2，异步客户端返回的是CompletableFuture，可以用allOf()一次性等待所有任务完成，代码更简洁高效：

// SDK v2的异步客户端初始化
KinesisAsyncClient kinesisClient = KinesisAsyncClient.create();
List<CompletableFuture<PutRecordResponse>> writeFutures = new ArrayList<>();

for (YourRecord dataRecord : recordsToSend) {
    PutRecordRequest request = PutRecordRequest.builder()
        .streamName("your-target-stream")
        .data(SdkBytes.fromByteArray(dataRecord.getContent()))
        .partitionKey(dataRecord.getPartitionKey())
        .build();
    
    CompletableFuture<PutRecordResponse> future = kinesisClient.putRecord(request);
    writeFutures.add(future);
}

// 等待所有异步任务完成
CompletableFuture.allOf(writeFutures.toArray(new CompletableFuture[0])).join();

// 可选：遍历结果处理成功/失败
for (CompletableFuture<PutRecordResponse> future : writeFutures) {
    try {
        PutRecordResponse response = future.get();
        System.out.println("写入成功，SequenceNumber: " + response.sequenceNumber());
    } catch (Exception e) {
        System.err.println("Kinesis写入失败: " + e.getMessage());
    }
}

SDK v2的异步客户端性能更好，CompletableFuture的API也更灵活，是官方推荐的用法。

方案3：用CountDownLatch控制等待（适合需要自定义回调的场景） #

如果需要在异步任务完成时做自定义逻辑（比如统计成功失败数），可以用CountDownLatch来跟踪任务数量：

AmazonKinesisAsync kinesisClient = AmazonKinesisAsyncClientBuilder.defaultClient();
// 初始化计数器，值等于要写入的记录数
CountDownLatch writeLatch = new CountDownLatch(recordsToSend.size());
AtomicInteger successCount = new AtomicInteger(0);
AtomicInteger failCount = new AtomicInteger(0);

for (YourRecord dataRecord : recordsToSend) {
    PutRecordRequest request = new PutRecordRequest()
        .withStreamName("your-target-stream")
        .withData(ByteBuffer.wrap(dataRecord.getContent()))
        .withPartitionKey(dataRecord.getPartitionKey());
    
    kinesisClient.putRecordAsync(request, new AsyncHandler<PutRecordRequest, PutRecordResult>() {
        @Override
        public void onSuccess(PutRecordRequest req, PutRecordResult res) {
            successCount.incrementAndGet();
            writeLatch.countDown();
        }

        @Override
        public void onError(Exception e) {
            failCount.incrementAndGet();
            System.err.println("写入失败: " + e.getMessage());
            writeLatch.countDown();
        }
    });
}

// 等待所有任务完成，设置超时时间避免无限阻塞
try {
    if (!writeLatch.await(30, TimeUnit.SECONDS)) {
        System.err.println("部分写入任务超时未完成");
    }
    System.out.println("写入统计：成功" + successCount.get() + "条，失败" + failCount.get() + "条");
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    System.err.println("等待写入任务时被中断");
}

这种方式可以更灵活地处理每个任务的结果，还能设置超时时间，避免Lambda被无限挂起。

额外注意事项 #

• Lambda超时时间：一定要设置足够长的Lambda超时时间，确保能覆盖所有异步写入的耗时（比如100条记录每条耗时100ms，至少设置5秒以上的超时），不然即使你等待，Lambda超时还是会被强制终止。
• 异常处理：不要忽略异步任务的异常，否则失败的请求不会被察觉，严重的话会导致数据丢失。
• SDK版本：尽量迁移到AWS SDK v2，v1的AmazonKinesisAsyncClient已经被标记为废弃，v2的异步客户端性能和扩展性更好。

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