小编Luc*_*ire的帖子

我在使用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();