TelpoPushThirdSsl/TelpoPush.Service/Mq/Kafka/MessageProducer.cs

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

namespace TelpoPush.Service.Mq.Kafka
{
    /// <summary>
    /// 消息生产者
    /// </summary>
    public class MessageProducer
    {
        private readonly ILogger<MessageProducer> _logger;
        private readonly ServiceConfig _configService;
        private readonly IProducer<Null, string> _producer;

        public MessageProducer(ILogger<MessageProducer> logger, IOptions<ServiceConfig> optConfigService)
        {
            _logger = logger;
            _configService = optConfigService.Value;

            var config = new ProducerConfig
            {
                BootstrapServers = _configService.KafkaBootstrapServers,
                EnableIdempotence = true,
                Acks = Acks.All,
                //LingerMs=5000,
                //BatchNumMessages =1000,
                //BatchSize=32768,
                //CompressionType= CompressionType.Lz4,
                MessageSendMaxRetries = 3
            };
            _producer = new ProducerBuilder<Null, string>(config).Build();
        }

        public Headers CreateHeader(Dictionary<string, int> pair = null)
        {
            if (pair == null)
            {
                return null;
            }
            else
            {
                Headers headers = new Headers();

                foreach (var item in pair)
                {
                    headers.Add(item.Key, BitConverter.GetBytes(item.Value));
                }
                return headers;
            }
        }

        public async Task ProduceAsync(List<TopicModel> topic, object message)
        {
            try
            {
                foreach (var item in topic)
                {
                    //  producer = new ProducerBuilder<Null, string>(config).Build();
                    await _producer.ProduceAsync(item.Topic, new Message<Null, string>
                    {
                        Headers = item.Headers,
                        Value = JsonConvert.SerializeObject(message)
                    });
                }
            }
            catch (ProduceException<Null, string> e)
            {
                _logger.LogError($"推送到kafka失败，topic: {topic}，\n message:{JsonConvert.SerializeObject(message)}: \n{e.Error.Reason}");
            }
        }



    }
}