AWS Kinesis NextShardIterator 永远不会为空

Question

上下文：我正在尝试使用 API 引用从 Kinesis 流中获取记录。我正在使用 .Net Core（3.1 版本）。

我正在使用 API 将数据写入 Kinesis Stream。这个问题没有任何问题。但是我在读取数据时遇到了一些问题。我将 getRecord 方法放在一个 do-while 循环中。 while 的条件是 nextShardIterator 值是否为空？但是这个值永远不会为空。我无法打破循环。

一些答案​​包括以下短语： "NextShardIterator 分片中开始顺序读取数据记录的下一个位置。如果设置为 null，则分片已关闭，请求的迭代器不再返回任何数据。"

我只有 1 个流和 1 个分片。我放了2条记录。之后，我执行阅读方法。它正在获取这些记录。但在那之后，即使记录被消耗，nextShardIterator 也永远不会为空。

• 列出流（我只有 1 个流）
• 描述流并获取分片（我只有 1 个分片）
• 获取 TRIM_HORIZON 类型的 ShardIterator
• GetRecord 在循环中 (while{if nextShardIterator != null})
• 找到一条记录后，我正在尝试使用 AFTER_SEQUENCE 分片迭代器
• 如果无法获取任何记录，我正在尝试使用 LATEST 分片迭代器

putRecord 代码是这样的：

public async Task<ResponseModel> PutRecord(string orderFlow, string documentId)
        {
            byte[] bytedata = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(orderFlow));
            using MemoryStream memoryStream = new MemoryStream(bytedata);
            var putRecordRequest = new PutRecordRequest();
            putRecordRequest.StreamName = myStreamName;
            putRecordRequest.PartitionKey = "partition" + documentId;
            putRecordRequest.Data = memoryStream;
            try
            {
                var putRecordResponse = await kinesisClient.PutRecordAsync(putRecordRequest);
                return new ResponseModel
                {
                    Data = putRecordResponse,
                    Status = ResponseStatus.Success,
                    Message = "Successfully put record!"
                };
            }
            catch (Exception e)
            {
                return new ResponseModel
                {
                    Data = null,
                    Status = ResponseStatus.Error,
                    Message = "PutRecord Error: " + e.Message
                };
            }
        }

但是我在读取数据时遇到了一些问题。 GetRecord 代码是这样的：

public async Task<ResponseModel> GetRecords()
        {
            var listStreams = await ListStreams();
            if (listStreams.StreamNames.Count == 0)
            {
                return new ResponseModel
                {
                    Data = null,
                    Status = ResponseStatus.Warning,
                    Message = "Do not have any Stream!"
                };
            }
            myStreamName = listStreams.StreamNames[0];

            var describeStreams = await DescribeStream(listStreams.StreamNames[0]);
            if (describeStreams.StreamDescription.StreamStatus != "ACTIVE")
            {
                return new ResponseModel
                {
                    Data = null,
                    Status = ResponseStatus.Warning,
                    Message = "Stream status: " + describeStreams.StreamDescription.StreamStatus
                };
            }

            var shards = describeStreams.StreamDescription.Shards;
            if (shards.Count == 0)
            {
                return new ResponseModel
                {
                    Data = null,
                    Status = ResponseStatus.Warning,
                    Message = "Do not have any Shard (or data)!"
                };
            }
            var shardId = shards[0].ShardId;

            var shardIterator = await GetShardIterator(shardId);
            if (string.IsNullOrWhiteSpace(shardIterator.ShardIterator))
            {
                return new ResponseModel
                {
                    Data = null,
                    Status = ResponseStatus.Warning,
                    Message = "ShardIterator is null or empty!"
                };
            }

            var getRecords = await GetRecords(shardIterator.ShardIterator);
            Console.WriteLine("First Iterator: " + shardIterator);
            var dataList = new List<Record>();

            do
            {
                if (getRecords.Records.Count == 0)
                {
                    Console.WriteLine("Records are empty!");
                    var nextShardIterator = GetShardLatest(shardId).Result.ShardIterator;
                    getRecords = await GetRecords(nextShardIterator);
                    Console.WriteLine("Latest Iterator: " + nextShardIterator);
                }
                else
                {
                    Console.WriteLine("We have records!");
                    foreach (var record in getRecords.Records)
                    {
                        dataList.Add(record);
                    }
                    var nextShardIterator = GetShardIteratorWithSequence(shardId, getRecords.Records[getRecords.Records.Count-1].SequenceNumber).Result.ShardIterator;
                    Console.WriteLine("AfterSequence Iterator: " + nextShardIterator);
                    getRecords = await GetRecords(nextShardIterator);
                }
            } while (getRecords.NextShardIterator != null);

            return new ResponseModel
            {
                Data = dataList,
                Status = ResponseStatus.Success,
                Message = "Successfull"
            };
        }

Answer 1

Kinesis 流是一个潜在的无限记录序列，可以由多个生产者随时添加。因此，打开流的分片迭代器永远不会为空。

如果您想在到达流的“结尾”时跳出循环，请查看GetRecords 响应中的MillisBehindLatest 字段。引用文档：

值为零表示记录处理已赶上，此时没有要处理的新记录。

但请注意，可能随时添加新记录。如果您确实跳出循环，请务必从您处理的最后一条记录中保存返回的 SequenceNumber，以便您可以从上次中断的地方继续。