zszq-trs/YLErpDAL/Helpers/KafkaProduceHelper.cs

using Confluent.Kafka;
using Microsoft.Extensions.Options;
using System.Linq;
using YLErp.Abstract;
using YLErp.Model;

namespace YLErp.Helpers
{
    public class KafkaProduceHelper: IKafkaProduce
    {
        private IProducer<string, string> _producer;
        private readonly IYcLogger _logger;
        public KafkaProduceHelper(IOptions<KafkaConfig> _kafkaConfig) {
            _logger = LogFactory.GetLogger("KafkaProduceHelper");
            var kfkconfig = _kafkaConfig.Value;
            ProducerConfig config = new ProducerConfig();
            config.BootstrapServers = kfkconfig.BootstrapServers;
            config.Acks = (Acks)kfkconfig.Acks;
            config.EnableIdempotence = kfkconfig.EnableIdempotence;
            config.MaxInFlight = kfkconfig.MaxInFlight;
            config.CompressionType = (CompressionType)kfkconfig.CompressionType;
            config.MessageTimeoutMs = kfkconfig.MessageTimeoutMs;
            _producer = new ProducerBuilder<string, string>(config).Build(); // 初始化生产者对象
        }
        public KafkaProduceHelper()
        {
            _logger = LogFactory.GetLogger("KafkaProduceHelper");
            var bootstrapServers = Environment.GetEnvironmentVariable("KafkaConfig_BootstrapServers");
            var batchSize = Environment.GetEnvironmentVariable("KafkaConfig_BatchSize");
            var acks = Environment.GetEnvironmentVariable("KafkaConfig_Acks");
            var enableIdempotence = Environment.GetEnvironmentVariable("KafkaConfig_EnableIdempotence");
            var maxInFlight = Environment.GetEnvironmentVariable("KafkaConfig_MaxInFlight");
            var compressionType = Environment.GetEnvironmentVariable("KafkaConfig_CompressionType");
            var messageTimeoutMs = Environment.GetEnvironmentVariable("KafkaConfig_MessageTimeoutMs");
            ProducerConfig config = new ProducerConfig();
            config.BootstrapServers = bootstrapServers;
            if (!string.IsNullOrEmpty(batchSize))
            {
                config.BatchSize = Convert.ToInt32(batchSize);
            }
            if (!string.IsNullOrEmpty(acks))
            {
                config.Acks = (Acks)(Convert.ToInt32(acks));
            }
            if (!string.IsNullOrEmpty(acks))
            {
                config.EnableIdempotence = Convert.ToBoolean(enableIdempotence);
            }
            if (!string.IsNullOrEmpty(maxInFlight))
            {
                config.MaxInFlight = Convert.ToInt32(maxInFlight);
            }
            if (!string.IsNullOrEmpty(compressionType))
            {
                config.CompressionType = (CompressionType)(Convert.ToInt32(compressionType));
            }
            if (!string.IsNullOrEmpty(messageTimeoutMs))
            {
                config.MessageTimeoutMs = Convert.ToInt32(messageTimeoutMs);
            }
            _producer = new ProducerBuilder<string, string>(config).Build(); // 初始化生产者对象
        }

        public void Produce(string topic,string message)
        {
            try
            {
                ProduceCore(topic, null, message);
            }
           catch (Exception ex)
            {
                _logger.Error($"Topic:{topic} send failed",ex);
            }
        }

        public void Produce(string topic, string key, string message)
        {
            ProduceCore(topic, key, message);
        }

        private void ProduceCore(string topic, string key, string message)
        {
            var kafkaMessage = new Message<string, string>
            {
                Key = key,
                Value = message
            };
            _producer.ProduceAsync(topic, kafkaMessage).GetAwaiter().GetResult();
        }
    }
}