配置kafka net停止发送最新消息

3z6pesqy 于 2021-06-08 发布在 Kafka

关注(0)|答案(1)|浏览(392)

我在带有kafka net插件的red hat vm上使用kafka 0.8.1.1。如何配置我的消费者停止接收来自kafka的早期消息？
我的消费者代码：

var options = new KafkaOptions(new Uri("tcp://199.53.249.150:9092"), new Uri("tcp://199.53.249.151:9092"));

Stopwatch sp = new Stopwatch();
var router = new BrokerRouter(options);
var consumer = new Consumer(new ConsumerOptions("Test", router));

ThreadStart start2 = () =>
{
    while (true)
    {
        sp.Start();
        foreach (var message in consumer.Consume())
        {
            if (MessageDecoderReceiver.MessageBase(message.Value) != null)
            {
                PrintMessage(MessageDecoderReceiver.MessageBase(message.Value).ToString());
            }
            else
            {
                Console.WriteLine(message.Value);
            }
        }
        sp.Stop();
    }
};
var thread2 = new Thread(start2);
thread2.Start();

apache-kafka c#kafka-consumer-api

来源：https://stackoverflow.com/questions/29042132/configure-kafka-net-to-stop-sending-latest-messages

1条答案

按热度按时间

wsxa1bj11#

kafka net中的使用者当前不会自动跟踪正在使用的偏移量。您必须手动执行偏移跟踪。
要在kafka版本0.8.1中存储偏移量，请执行以下操作：

var commit = new OffsetCommitRequest
            {
                ConsumerGroup = consumerGroup,
                OffsetCommits = new List<OffsetCommit>
                            {
                                new OffsetCommit
                                    {
                                        PartitionId = partitionId,
                                        Topic = IntegrationConfig.IntegrationTopic,
                                        Offset = offset,
                                        Metadata = metadata
                                    }
                            }
            };

var commitResponse = conn.Connection.SendAsync(commit).Result.FirstOrDefault();

要将使用者设置为在特定偏移点开始导入，请执行以下操作：

var offsets = consumer.GetTopicOffsetAsync(IntegrationConfig.IntegrationTopic).Result
                    .Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray();

var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationTopic, router), offsets);

注意上面的代码将消费者设置为从日志的最后开始消费，实际上只接收新消息。

赞(0）回复(0）举报 2021-06-08

我来回答

配置kafka net停止发送最新消息

1条答案

相关问题

热门标签

最新问答