TelpoPushPosition/TelpoPush.Position.Worker/Service/Mq/KafkaService.cs

using Confluent.Kafka;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using TelpoPush.Position.Worker.Models.Config;

namespace TelpoPush.Position.Worker.Service.Mq
{
    public class KafkaService : IKafkaService
    {
        private readonly ConsumerConfig _consumerConfig;
        private readonly IHostEnvironment env;
        private readonly ILogger<KafkaService> logger;
        private readonly ServiceConfig _configService;
        public KafkaService(ILogger<KafkaService> _logger, IHostEnvironment _env, IOptions<ServiceConfig> optConfigService)
        {
            _configService = optConfigService.Value;
            env = _env;
            logger = _logger;

            _consumerConfig = new ConsumerConfig
            {
                BootstrapServers = _configService.KafkaBootstrapServers,
                GroupId = _configService.KafkaGroupId,
                EnableAutoCommit = false,
                StatisticsIntervalMs = 5000,
                SessionTimeoutMs = 6000,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnablePartitionEof = true,
                CancellationDelayMaxMs=1
                
            };
        }

        public async Task SubscribeAsync(Action<string, string, Headers> messageFunc, CancellationToken cancellationToken)
        {
            List<string> topics = _configService.KafkaTopics;
            using (var consumer = new ConsumerBuilder<Ignore, string>(_consumerConfig)
                                 .SetErrorHandler((_, e) =>
                                 {
                                     Console.WriteLine($"Error: {e.Reason}");
                                     logger.LogError($"Error: {e.Reason}");
                                 })
                                 .SetStatisticsHandler((_, json) =>
                                 {
                                     Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} > 消息监听中..");
                                     logger.LogInformation($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} > 消息监听中..");
                                 })
                                 .SetPartitionsAssignedHandler((c, partitions) =>
                                 {
                                     string partitionsStr = string.Join(", ", partitions);
                                     logger.LogInformation($" - 分配的 kafka 分区: {partitionsStr}");
                                 })
                                 .SetPartitionsRevokedHandler((c, partitions) =>
                                 {
                                     string partitionsStr = string.Join(", ", partitions);
                                     Console.WriteLine($" - 回收了 kafka 的分区: {partitionsStr}");
                                     logger.LogInformation($" - 回收了 kafka 分区: {partitionsStr}");
                                 })
                                 .Build())
            {

                consumer.Subscribe(topics);
                try
                {
                    while (true)
                    {
                        try
                        {
                            var consumeResult = consumer.Consume(cancellationToken);
                            int DataType = -1, AlarmType = -1, OperType = -1;
                            foreach (var item in consumeResult?.Headers)
                            {
                                if (item.Key == KafkaHeader.DataType)
                                    DataType = BitConverter.ToInt32(item.GetValueBytes(), 0);
                                else if (item.Key == KafkaHeader.AlarmType)
                                    AlarmType = BitConverter.ToInt32(item.GetValueBytes(), 0);
                                else if (item.Key == KafkaHeader.OperType)
                                    OperType = BitConverter.ToInt32(item.GetValueBytes(), 0);
                            }
                            var Headers = new { DataType, AlarmType, OperType };
                            logger.LogInformation($"Consumed  topic '{consumeResult.Topic}', message '{consumeResult.Message?.Value}' ,  headers '{JsonConvert.SerializeObject(Headers)}', at: '{consumeResult?.TopicPartitionOffset}'.");
                            if (consumeResult.IsPartitionEOF)
                            {
                                logger.LogInformation($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} 已经到底了：{consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
                                Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} 已经到底了：{consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
                                continue;
                            }

                            string messageResult = null;
                            Headers headers = null;
                            try
                            {
                                messageResult = consumeResult.Message.Value;
                                headers = consumeResult.Message.Headers;
                            }
                            catch (Exception ex)
                            {
                                var errorMessage = $" - {DateTime.Now:yyyy-MM-dd HH:mm:ss}【Exception 消息反序列化失败，Value：{consumeResult.Message.Value}】 ：{ex.StackTrace?.ToString()}";
                                Console.WriteLine(errorMessage);
                                logger.LogError(errorMessage);
                                messageResult = null;
                            }
                            if (!string.IsNullOrEmpty(messageResult)/* && consumeResult.Offset % commitPeriod == 0*/)
                            {
                                string topic = consumeResult.Topic;
                                messageFunc(topic, messageResult, headers);

                                try
                                {
                                    consumer.Commit(consumeResult);
                                }
                                catch (KafkaException e)
                                {
                                    Console.WriteLine(e.Message);
                                }
                            }
                        }
                        catch (ConsumeException e)
                        {
                            logger.LogError($"Consume error: {e.Error.Reason}");
                            Console.WriteLine($"Consume error: {e.Error.Reason}");
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    logger.LogError("Closing consumer.");
                    Console.WriteLine("Closing consumer.");
                    consumer.Close();
                }
            }
            await Task.CompletedTask;
        }
    }
}