我正在使用AWS Kinesis进行生产者和消费者实验,但遇到的问题是尽管我们多次更改发送的数据对象,消费者仍然接收第一条消息(或记录)。此外,我们尝试了多种ShardIteratorType,但都没有起作用。最新的类型不产生任何结果,而其他所有类型仍会产生相同的原始记录。
using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading.Tasks;
using Amazon;
using Amazon.Internal;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
using BenchmarkRuleSetModel.Models;
using MongoDB.Driver;
using Newtonsoft.Json;
namespace ConsoleApp7
{
internal class Program
{
private static AmazonKinesisClient _client;
private static string _streamName;
static async Task ReadFromStream()
{
var kinesisStreamName = _streamName;
var describeRequest = new DescribeStreamRequest
{
StreamName = kinesisStreamName,
};
var describeResponse = await _client.DescribeStreamAsync(describeRequest);
var shards = describeResponse.StreamDescription.Shards;
foreach (var shard in shards)
{
var iteratorRequest = new GetShardIteratorRequest
{
StreamName = kinesisStreamName,
ShardId = shard.ShardId,
ShardIteratorType = ShardIteratorType.AT_TIMESTAMP,
Timestamp = DateTime.MinValue
};
var iteratorResponse = await _client.GetShardIteratorAsync(iteratorRequest);
var iteratorId = iteratorResponse.ShardIterator;
while (!string.IsNullOrEmpty(iteratorId))
{
var getRequest = new GetRecordsRequest
{
ShardIterator = iteratorId, Limit = 10000
};
var getResponse = await _client.GetRecordsAsync(getRequest);
var nextIterator = getResponse.NextShardIterator;
var records = getResponse.Records;
if (records.Count > 0)
{
Console.WriteLine("Received {0} records. ", records.Count);
foreach (var record in records)
{
var json = Encoding.UTF8.GetString(record.Data.ToArray());
Console.WriteLine("Json string: " + json);
}
}
iteratorId = nextIterator;
}
}
}
private static async Task<string> Produce()
{
var data = new
{
Message = "Hello world!",
Author = "Amir"
};
//convert to byte array in prep for adding to stream
var oByte = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(data));
using (var ms = new MemoryStream(oByte))
{
//create put request
var requestRecord = new PutRecordRequest
{
StreamName = _streamName,
PartitionKey = Guid.NewGuid().ToString(),
Data = ms
};
//list name of Kinesis stream
//give partition key that is used to place record in particular shard
//add record as memorystream
//PUT the record to Kinesis
var response = await _client.PutRecordAsync(requestRecord);
return response.SequenceNumber;
}
}
static void Main(string[] args)
{
_client = new AmazonKinesisClient("ExampleKey", "ExampleSecret", RegionEndpoint.EUWest2);
_streamName = "SomeStream";
Produce().Wait();
ReadFromStream().Wait();
}
}
}