【Kafka】基于Windows环境的Kafka有关环境(scala+zookeeper+kafka+可视化工具)搭建、以及使用.NET环境开发的案例代码与演示

/// <summary>
    /// Description: Kafka生产者发布服务
    /// CreateTime: 2022/1/21 19:35:27
    /// Author: Wesky
    /// </summary>
    public class PublishService: IPublishService
    {
        public async Task PublishAsync<TMessage>(string broker, string topicName, TMessage message) where TMessage : class
        {
            var config = new ProducerConfig
            {
                BootstrapServers = broker, // kafka服务集群，例如  "192.168.0.1:9092,192.168.0.2:9092"   或者单机 "192.168.0.1:9092"
                Acks = Acks.All,
                MessageSendMaxRetries = 3, // 发送失败重试的次数
            };
            using (var producer = new ProducerBuilder<string, string>(config).Build())
            {
                try
                {
                    string data = Newtonsoft.Json.JsonConvert.SerializeObject(message);
                    var sendData = new Message<string, string> { Key = Guid.NewGuid().ToString("N"), Value = data};
                    var report =  await producer.ProduceAsync(topicName, sendData);
                    Console.WriteLine($"消息 >>>>>： {data} \r\n发送到：{report.TopicPartitionOffset}");
                }
                catch (ProduceException<string, string> ex)
                {
                    Console.WriteLine($"消息发送失败>>>>>：\r\n Code= {ex.Error.Code} >>> \r\nError= {ex.Message}");
                }
            }
        }
    }

/// <summary>
    /// Description: kafka 消费者订阅服务
    /// CreateTime: 2022/1/21 19:36:25
    /// Author: Wesky
    /// </summary>
    public class SubscribeService: ISubscribeService
    {
        /// <summary>
        /// 消费者服务核心代码
        /// </summary>
        /// <typeparam name="TMessage"></typeparam>
        /// <param name="config">消费者配置信息</param>
        /// <param name="topics">主题集合</param>
        /// <param name="func"></param>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public async Task SubscribeAsync<TMessage>(ConsumerConfig config, IEnumerable<string> topics,  Action<TMessage> func, CancellationToken cancellationToken) where TMessage : class
        {
            const int commitPeriod = 1;
            using (var consumer = new ConsumerBuilder<Ignore, string>(config)
             .SetErrorHandler((_, e) =>
             {
                 Console.WriteLine($"消费错误 >>>>>: {e.Reason}");
             })
             .SetStatisticsHandler((_, json) =>
             {
                 Console.WriteLine($"************************************************");
             })
             .SetPartitionsAssignedHandler((c, partitionList) =>
             {
                 string partitions = string.Join(", ", partitionList);
                 Console.WriteLine($"分配的分区 >>>>> : {partitions}");
             })
             .SetPartitionsRevokedHandler((c, partitionList) =>
             {
                 string partitions = string.Join(", ", partitionList);
                 Console.WriteLine($"回收的分区 >>>>> : {partitions}");
             })
             .Build())
            {
                consumer.Subscribe(topics);
                try
                {
                    while (true)
                    {
                        try
                        {
                            var consumeResult = consumer.Consume(cancellationToken);
                            if (consumeResult.IsPartitionEOF)
                            {
                                continue;
                            }
                            if(consumeResult?.Offset % commitPeriod == 0){
                                try
                                {
                                    var result = JsonConvert.DeserializeObject<TMessage>(consumeResult.Message?.Value);
                                    func(result); // 消费消息
                                }
                                catch (Exception ex)
                                {
                                    Console.WriteLine($"消费业务处理失败: {ex.Message}");
                                }
                                try
                                {
                                    consumer.Commit(consumeResult);  // 手动提交
                                    Console.WriteLine($"消费者消费完成，已提交 ");
                                }
                                catch (KafkaException e)
                                {
                                    Console.WriteLine($"提交错误 >>>>> : {e.Error.Reason}");
                                }
                            }
                        }
                        catch (ConsumeException e)
                        {
                            Console.WriteLine($"消费错误>>>>> : {e.Error.Reason}");
                        }
                    }
                }
                catch (Exception e)
                {
                    Console.WriteLine($"其他错误 >>>>> :{e.Message}");
                    consumer.Close();
                }
            }
            await Task.CompletedTask;
        }
    }

[Route("api/[controller]/[action]")]
    [ApiController]
    public class ProducerController : ControllerBase
    {
        IPublishService _service = null;
        public ProducerController(IPublishService publishService)
        {
            _service = publishService;
        }
        [HttpPost]
        public IActionResult SendMessage(string broker,string topicName,string message)
        {
            _service.PublishAsync(broker, topicName, message);
            return Ok();
        }
    }