我刚开始使用 Kinesis,其 API 可在此处获得
我用它把100 条记录推送到kinesis
for (int j = 0; j < 100; j++) {
PutRecordRequest putRecordRequest = new PutRecordRequest();
putRecordRequest.setStreamName(myStreamName);
putRecordRequest.setData(ByteBuffer.wrap(data.getBytes()));
putRecordRequest.setPartitionKey(String.format("partitionKey-%d", j));
PutRecordResult putRecordResult = kinesisClient.putRecord(putRecordRequest);
System.out.println("Successfully putrecord, partition key : " + putRecordRequest.getPartitionKey()
+ ", ShardID : " + putRecordResult.getShardId() + ", Sequence No : "+ putRecordResult.getSequenceNumber());
}
现在我想获取推送的记录数。为此,我正在使用这个:
Iterator<Shard> shardIterator = getTotalShardsIterator();//Implemented and giving perfectly all the shards.....
现在使用上面的迭代器,我得到的计数是:
.....
while (shardIterator.hasNext()) {
Shard shard = shardIterator.next();
String shardId = shard.getShardId();
int datacount = getDataCount(shardId, myStreamName);
totalStreamDataCount+= datacount;
System.out.println("Data Count for Shard " + shardId + " is : " + datacount);
}
.....
这是我的函数getDataCount(shardId, myStreamName)
public static int getDataCount(String shardId, String streamName) {
int dataCount = 0;
String shardIterator;
GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest();
getShardIteratorRequest.setStreamName(streamName);
getShardIteratorRequest.setShardId(shardId);
getShardIteratorRequest.setShardIteratorType(ShardIteratorType.TRIM_HORIZON);
GetShardIteratorResult getShardIteratorResult = kinesisClient.getShardIterator(getShardIteratorRequest);
shardIterator = getShardIteratorResult.getShardIterator();
GetRecordsRequest getRecordsRequest = new GetRecordsRequest();
getRecordsRequest.setShardIterator(shardIterator);
getRecordsRequest.setLimit(1000);
GetRecordsResult getRecordsResult = kinesisClient.getRecords(getRecordsRequest);
List<Record> records = getRecordsResult.getRecords();
if(!records.isEmpty() && records.size() > 0){
dataCount = records.size();
Iterator<Record> iterator = records.iterator();
while(iterator.hasNext()) {
Record record = iterator.next();
byte[] bytes = record.getData().array();
String recordData = new String(bytes);
System.out.println("Shard Id. :"+shardId+"Seq. No. is : "+" Record data :"+recordData);
}
}
return dataCount;
}
但是这段代码每次运行时都会给出不匹配的结果,比如有时它会显示 81 有时 91
请对此有所了解.. :)