如何使用上次保存的偏移量中的kafka消息

Question

我是Kafka的新手，一直在尝试实现一个消费者。下面是我的场景

1. 启动消费者应用程序
2. 产生来自生产者的消息。这些消息被消费者消费
3. 停止消费者并再次生成消息。当我启动消费者时，在消费者被停止时发布的消息不会被读取

虽然auto.offset.commit=最早会消耗消息，但它会消耗发布到主题的所有消息。我想只消耗那些在消费者关闭时发布的消息。

var options = new KafkaOptions(new Uri(kafkaUri));
var router = new BrokerRouter(options);
var consumer = new Consumer(new ConsumerOptions(kafkaTopic, router));
var offset = consumer.GetTopicOffsetAsync(kafkaTopic, 100000).Result;
var t = from x in offset select new OffsetPosition(x.PartitionId, x.Offsets.Max());
consumer.SetOffsetPosition(t.ToArray());

foreach (Message msg in consumer.Consume()) {
    string kafkaResponse = System.Text.Encoding.UTF8.GetString(msg.Value);

    Console.WriteLine("PickList Json : " + kafkaResponse);

    offsetCommitRequest.Offset = msg.Meta.Offset;
    offsetCommitRequest.PartitionId = msg.Meta.PartitionId;
    offsetCommitRequest.Topic = kafkaTopic;
    offsetCommitRequest.Metadata = "CommitOffset";
    var offsetCommitResponse = await _kafkaPublishService.SetOffsetvalue(kafkaUri, kafkaTopic, consumerGroup, offsetCommitRequest);
}

Answer 1

您应该重新启动您的消费者，而无需设置任何手动偏移量。意味着你不应该这样做：

var offset = consumer.GetTopicOffsetAsync(kafkaTopic, 100000).Result;
var t = from x in offset select new OffsetPosition(x.PartitionId, x.Offsets.Max());
consumer.SetOffsetPosition(t.ToArray());

您的消费者将从上次promise的地方赶上

亚尼克