.Net Core 集成 Kafka的步骤(2)

There are a couple of additional benefits of using the Produce method. First, notification of message delivery (or failure) is strictly in the order of broker acknowledgement. With ProduceAsync, this is not the case because Tasks may complete on any thread pool thread. Second, Produce is more performant because there is unavoidable overhead in the higher level Task based API.

消费者

static void Main(string[] args) { Console.WriteLine("Hello World kafka consumer !"); var config = new ConsumerConfig { BootstrapServers = "192.168.0.117:9092", GroupId = "foo", AutoOffsetReset = AutoOffsetReset.Earliest }; var cancel = false; using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build()) { var topic = "test"; consumer.Subscribe(topic); while (!cancel) { var consumeResult = consumer.Consume(CancellationToken.None); Console.WriteLine($"Consumer message: { consumeResult.Message.Value} topic: {consumeResult.Topic} Partition: {consumeResult.Partition}"); } consumer.Close(); } }

消费者的演示代码同样很简单。我们需要指定groupId，然后订阅topic。使用ConsumerBuilder构造一个consumer，然后调用Consume方法进行消费就可以。
注意：
这里默认是自动commit消费。你也可以根据情况手动提交commit。

运行一下

我们运行一个生产者进程，按照500ms的速度生产消息。运行三个consumer进行消费，可以看到消息被均匀的推送到三个consumer上去。

总结

以上简单的介绍了kafka的背景、安装方法、使用场景。还简单演示了如何使用.net来操作kafka。它可以当作流式计算平台来使用，也可以当作传统的消息队列使用。它当前非常流行，网上的资料也多如牛毛。官方也提供了简单易用的.net sdk ，为.net 平台集成kafka提供了便利。