我最近开始使用.NETCore开发kafka流媒体应用程序。我遵循了教程：https://medium.com/@srigumm/building-realtime-streaming-applications-using-net-core-and-kafka-ad45ed081b31。
我已经构建了一个基本的生产者-消费者应用程序，生产者在其中获取输入数据并将其推入Kafka主题。使用者可以订阅主题并使用其中的数据。我也可以通过使用一个新的生产者将这些数据推到另一个主题中。但我不能做的是初始化多个消费者从同一主题消费。
在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。但我不知道在哪里，怎么做。