以下是使用FirehoseAsyncClient和PutRecordBatch方法将数据记录放入Firehose流的Java代码示例:
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.firehose.FirehoseAsyncClient;
import software.amazon.awssdk.services.firehose.model.PutRecordBatchRequest;
import software.amazon.awssdk.services.firehose.model.PutRecordBatchResponse;
import software.amazon.awssdk.services.firehose.model.Record;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
public class FirehoseExample {
public static void main(String[] args) {
// 设置AWS区域
Region region = Region.US_EAST_1;
// 创建FirehoseAsyncClient
FirehoseAsyncClient firehoseClient = FirehoseAsyncClient.builder()
.region(region)
.build();
// 准备要发送到Firehose流的数据记录
List records = new ArrayList<>();
Record record1 = Record.builder()
.data(SdkBytes.fromUtf8String("Record 1"))
.build();
Record record2 = Record.builder()
.data(SdkBytes.fromUtf8String("Record 2"))
.build();
records.add(record1);
records.add(record2);
// 创建PutRecordBatch请求
PutRecordBatchRequest request = PutRecordBatchRequest.builder()
.deliveryStreamName("your_delivery_stream_name")
.records(records)
.build();
// 异步发送数据记录到Firehose流
CompletableFuture future = firehoseClient.putRecordBatch(request);
future.whenComplete((response, exception) -> {
if (response != null) {
System.out.println("成功发送了 " + response.requestResponses().size() + " 条记录");
} else {
System.out.println("发送数据记录时出现错误: " + exception.getMessage());
}
firehoseClient.close();
});
// 等待异步操作完成
future.join();
}
}
请确保将your_delivery_stream_name替换为您自己的Firehose流的名称。