如何从kafka流中添加多个消费者来消费？

Question

我最近开始使用 .NET 核心开发 Kafka 流应用程序。我已按照教程进行操作：https://medium.com/@srigumm/building-realtime-streaming-applications-using-net-core-and-kafka-ad45ed081b31。

我已经构建了一个基本的生产者-消费者应用程序，生产者在其中获取输入数据并将其推送到 kafka-topic 中。消费者可以订阅主题并使用其中的数据。我还可以通过使用新的生产者将这些数据推送到另一个主题。但我无法做的是初始化多个消费者从同一个主题消费。

在 appsettings.json 中：

  "consumer": {
    "bootstrapservers": "localhost:9092", //specify your kafka broker address
    "groupid": "csharp-consumer",
    "enableautocommit": true,
    "statisticsintervalms": 5000,
    "sessiontimeoutms": 6000,
    "autooffsetreset": 0,
    "enablepartitioneof": true,
    "SaslMechanism": 0, //0 for GSSAPI
    //"SaslKerberosKeytab":"filename.keytab", //specify your keytab file here
    "SaslKerberosPrincipal": "youralias@DOMAIN.COM", //specify your alias here
    "SaslKerberosServiceName": "kafka"
    //"SaslKerberosKinitCmd":"kinit -k -t %{sasl.kerberos.keytab} %{sasl.kerberos.principal}"
  },

processOrderServices.cs：

namespace Api.Services
{

    public class ProcessOrdersService : BackgroundService
    {
        private readonly ConsumerConfig consumerConfig;
        private readonly ProducerConfig producerConfig;
        //----------------------
        //private readonly ConsumerConfig consumerConfig2;
        public ProcessOrdersService(ConsumerConfig consumerConfig, ProducerConfig producerConfig)
        {
            this.producerConfig = producerConfig;
            this.consumerConfig = consumerConfig;
            //----------------------
            //this.consumerConfig2 = consumerConfig;
        }
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            Console.WriteLine("OrderProcessing Service Started\n\n");

            while (!stoppingToken.IsCancellationRequested)
            {
                var consumerHelper = new ConsumerWrapper(consumerConfig, "orderrequests");
                //var consumerHelper2 = new ConsumerWrapper(consumerConfig, "orderrequests");
                string orderRequest = consumerHelper.readMessage();
                //consumerHelper2.DisplayMessage();

                //Deserilaize 
                OrderRequest order = JsonConvert.DeserializeObject<OrderRequest>(orderRequest);
                //TODO:: Process Order
                Console.WriteLine($"Info: OrderHandler => Processing the order for {order.productname}\n\n");
                order.status = OrderStatus.COMPLETED;

                //Write to ReadyToShip Queue

                var producerWrapper = new ProducerWrapper(producerConfig,"readytoship");
                await producerWrapper.writeMessage(JsonConvert.SerializeObject(order));
                //--------------------------
               // var consumerHelper2 = new ConsumerWrapper(consumerConfig2, "orderrequests");
                //string processedOrder = consumerHelper2.readMessage();
                //OrderRequest order2 = JsonConvert.DeserializeObject<OrderRequest>(processedOrder);
                //Console.WriteLine($"Info: OrderHandler => Delivered the order for {order2.productname}\n\n");
                //order2.status = OrderStatus.DELIVERED;
                //----------------------------
            }
        }
    }
} 

ConsumerWrapper.cs：

namespace Api
{

    public class ConsumerWrapper
    {
        private string _topicName;
        private ConsumerConfig _consumerConfig;
        private Consumer<string,string> _consumer;
        private static readonly Random rand = new Random();
        public ConsumerWrapper(ConsumerConfig config,string topicName)
        {
            this._topicName = topicName;
            this._consumerConfig = config;
            this._consumer = new Consumer<string,string>(this._consumerConfig);
            this._consumer.Subscribe(topicName);
        }
        public string readMessage(){
            var consumeResult = this._consumer.Consume();
            return consumeResult.Value;
        }
        public void DisplayMessage()
        {
            var consumeResult = this._consumer.Consume();
            Console.WriteLine(consumeResult.Value);
            Console.WriteLine($"Info: OrderHandler => Delivered the order for {consumeResult.Value}\n\n");
            return;
        }
    }
} 

我希望能够多次调用 Consumer 类并能够阅读同一主题。我知道需要创建多个分区/组 ID 才能做到这一点。但我无法弄清楚在哪里以及如何做到这一点。

Answer 1

您可以通过在Kafka中使用组ID概念来做到这一点，只需为多个消费者使用相同的组ID，以避免重复消费来自同一主题的数据。