public static async Task Main()
        {
            var config = new ProducerConfig
            {
                BootstrapServers = "172.16.126.69:9092",
                ClientId= Dns.GetHostName(),
            };

            // Create a producer that can be used to send messages to kafka that have no key and a value of type string 
            using var producer  = new ProducerBuilder<string, string>(config).Build();

            var i = 0;
            while (true)
            {    
                // 输出内容到控制台
                Console.WriteLine("请输入Key");
                var key= Console.ReadLine();
                Console.WriteLine("请输入value");
                var value= Console.ReadLine();
                
                var message = new Message<string, string>
                {
                    Key = $"{key}",
                    Value = $"{value}"
                };

                // Send the message to our test topic in Kafka                
                var dr = await producer.ProduceAsync(new TopicPartition(topic: "test", partition: new Partition(0)), message);
            
                Console.WriteLine($"Produced message '{dr.Value}' to topic {dr.Topic}, partition {dr.Partition}, offset {dr.Offset}");
            }                                                                                              
        }

 class Program
    {
        static void Main()
        {
            var conf = new ConsumerConfig
            {
                GroupId = "test",
                BootstrapServers = "172.16.126.69:9092",
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableAutoCommit = false,
                //EnableAutoOffsetStore = false
                //SessionTimeoutMs = 1000 * 60 * 30, //30min
                //MaxPollIntervalMs = 1000 * 60 * 30,  //30min,30分钟不轮询kafak服务器将移除所在Group
            };

            using var consumer = new ConsumerBuilder<Ignore, string>(conf).Build();

            ////无法指定分区进行消费
            consumer.Subscribe("^admin");
            //var s = consumer.Assignment;

            //consumer.Assign(new TopicPartition("admin", new Partition(4)));

            //获取指定分区的当前偏移量
            //var unused = consumer.Position(new TopicPartition("", new Partition(0)));
            var cts = new CancellationTokenSource();
            Console.CancelKeyPress += (_, e) =>
            {
                e.Cancel = true;
                cts.Cancel();
            };

            try
            {
                while (true)
                {
                    //consumer.Assign(new TopicPartitionOffset("admin", new Partition(0),new Offset(391))); 
                    //consumer.Assign(new TopicPartition("admin", new Partition(0)));
                    var consumeResult = consumer.Consume(cts.Token);
                    if (consumeResult.Message.Headers.TryGetLastBytes("GameId", out byte[] lastHeader))
                    {
                        var header = Encoding.Default.GetString(lastHeader);
                        Console.WriteLine($"头为：{header}");
                    }

                    Console.WriteLine($"Consumed message '{consumeResult.Message.Value}' from topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}");

                    //consumer.StoreOffset(consumeResult);
                    //获取当前topic分区数
                    //var count = consumer.Assignment.Count;
                    //重置分区 offset，可实现 回读，跳过
                    //consumeResult.TopicPartitionOffset = new TopicPartitionOffset("admin", new Partition(0), new Offset(321));
                    try
                    {
                        //业务执行成功后可能kafka事务提交失败，故拿到消费数据后先判断消息是否消费过
                        //(方案：为每条消息加一个标识符，业务执行成功后标识符状态也要同时(事务)改为已消费，可解决重复消费问题)
                        
                        consumer.Assign(new TopicPartitionOffset(consumeResult.Topic, new Partition(consumeResult.Partition), new Offset(consumeResult.Offset+1)));//设置本地偏移量+1，可省略
                        consumer.Commit(consumeResult);//将本地偏移量+1提交到kafka服务端
                        //等价于手动ACK,即使不提交，但是本地offset始终会+1，kafka服务端不会+1，
                        //为了让本地offset与服务端offset一致，需手动重置本地TopicPartitionOffset并调用Assign()方法重置本地kafka数据
                        
                        //consumer.StoreOffset(consumeResult);
                    }
                    catch (KafkaException e)
                    {
                        Console.WriteLine($"Commit error: {e.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {
            }
            finally
            {
                consumer.Close();
            }
        }
    }

 using var consumer = new ConsumerBuilder<Ignore, string>(conf).Build();
  consumer.Assign(new TopicPartition("admin", new Partition(4)));//指定消费admin 主题中的4分区；指定分区消费后会变成广播模式，其他消费者依然会消费，不受此影响