[開源]基於Log4Net簡單實現KafkaAppender

背景

1. 基於以前基於Log4Net本地日誌服務簡單實現 實現本地日誌服務，可是隨着項目開發演進，本地日誌服務知足不了需求，譬如在預發佈環境或者生產環境，不可能讓開發人員登陸查看本地日誌文件分析。
2. Kafka+ELK日誌服務套件，能夠在線日誌服務能夠解決上述問題，而且提供豐富報表分析等等；
3. 具體源碼：MasterChief
4. Nuget：Install-Package MasterChief.DotNet.Core.KafkaLog
5. 歡迎Star，歡迎Issues；

源碼

1. 基於Log4Net來實現與kafka通信Appender

   public class KafkaAppender : AppenderSkeleton
    {
        #region Fields
   
        /// <summary>
        ///     Kafka 生產者
        /// </summary>
        private Producer _kafkaProducer;
   
        #endregion Fields
   
        #region Properties
   
        /// <summary>
        ///     Brokers
        /// </summary>
        public string Brokers { get; set; }
   
        /// <summary>
        ///     Topic
        /// </summary>
        public string Topic { get; set; }
   
        #endregion Properties
   
        #region Methods
   
        /// <summary>
        ///     Initialize the appender based on the options set
        /// </summary>
        /// <remarks>
        ///     <para>
        ///         This is part of the <see cref="T:log4net.Core.IOptionHandler" /> delayed object
        ///         activation scheme. The <see cref="M:log4net.Appender.AppenderSkeleton.ActivateOptions" /> method must
        ///         be called on this object after the configuration properties have
        ///         been set. Until <see cref="M:log4net.Appender.AppenderSkeleton.ActivateOptions" /> is called this
        ///         object is in an undefined state and must not be used.
        ///     </para>
        ///     <para>
        ///         If any of the configuration properties are modified then
        ///         <see cref="M:log4net.Appender.AppenderSkeleton.ActivateOptions" /> must be called again.
        ///     </para>
        /// </remarks>
        public override void ActivateOptions()
        {
            base.ActivateOptions();
            InitKafkaProducer();
        }
   
        /// <summary>
        ///     Subclasses of <see cref="T:log4net.Appender.AppenderSkeleton" /> should implement this method
        ///     to perform actual logging.
        /// </summary>
        /// <param name="loggingEvent">The event to append.</param>
        /// <remarks>
        ///     <para>
        ///         A subclass must implement this method to perform
        ///         logging of the <paramref name="loggingEvent" />.
        ///     </para>
        ///     <para>
        ///         This method will be called by <see cref="M:DoAppend(LoggingEvent)" />
        ///         if all the conditions listed for that method are met.
        ///     </para>
        ///     <para>
        ///         To restrict the logging of events in the appender
        ///         override the <see cref="M:PreAppendCheck()" /> method.
        ///     </para>
        /// </remarks>
        protected override void Append(LoggingEvent loggingEvent)
        {
            try
            {
                var message = GetLogMessage(loggingEvent);
                var topic = GetTopic(loggingEvent);
   
                _ = _kafkaProducer.SendMessageAsync(topic, new[] {new Message(message)});
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("KafkaProducer SendMessageAsync", ex);
            }
        }
   
        /// <summary>
        ///     Raises the Close event.
        /// </summary>
        /// <remarks>
        ///     <para>
        ///         Releases any resources allocated within the appender such as file handles,
        ///         network connections, etc.
        ///     </para>
        ///     <para>
        ///         It is a programming error to append to a closed appender.
        ///     </para>
        /// </remarks>
        protected override void OnClose()
        {
            base.OnClose();
            StopKafkaProducer();
        }
   
        private string GetLogMessage(LoggingEvent loggingEvent)
        {
            var builder = new StringBuilder();
            using (var writer = new StringWriter(builder))
            {
                Layout.Format(writer, loggingEvent);
   
                if (Layout.IgnoresException && loggingEvent.ExceptionObject != null)
                    writer.Write(loggingEvent.GetExceptionString());
   
                return writer.ToString();
            }
        }
   
        private string GetTopic(LoggingEvent loggingEvent)
        {
            return string.IsNullOrEmpty(Topic) ? Path.GetFileNameWithoutExtension(loggingEvent.Domain) : Topic;
        }
   
        /// <summary>
        ///     初始化Kafka 生產者
        /// </summary>
        private void InitKafkaProducer()
        {
            try
            {
                if (string.IsNullOrEmpty(Brokers)) Brokers = "http://localhost:9200";
   
                if (_kafkaProducer == null)
                {
                    var brokers = new Uri(Brokers);
                    var kafkaOptions = new KafkaOptions(brokers)
                    {
                        Log = new KafkaLog()
                    };
                    _kafkaProducer = new Producer(new BrokerRouter(kafkaOptions));
                }
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("InitKafkaProducer", ex);
            }
        }
   
        /// <summary>
        ///     中止生產者
        /// </summary>
        private void StopKafkaProducer()
        {
            try
            {
                _kafkaProducer?.Stop();
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("StopKafkaProducer", ex);
            }
        }
   
        #endregion Methods
    }

2. 基於以前定義接口，來實現kafkaLogService

   public sealed class KafkaLogService : ILogService
   {
       #region Constructors
   
       /// <summary>
       ///     Initializes the <see cref="FileLogService" /> class.
       /// </summary>
       static KafkaLogService()
       {
           KafkaLogger = LogManager.GetLogger(KafkaLoggerName);
       }
   
       #endregion Constructors
   
       #region Fields
   
       /// <summary>
       ///     Kafka logger name
       /// </summary>
       public const string KafkaLoggerName = "KafkaLogger";
   
       /// <summary>
       ///     Kafka logger
       /// </summary>
       public static readonly ILog KafkaLogger;
   
       #endregion Fields
   
       #region Methods
   
       /// <summary>
       ///     Debug記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       public void Debug(string message)
       {
           if (KafkaLogger.IsDebugEnabled) KafkaLogger.Debug(message);
       }
   
       /// <summary>
       ///     Debug記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       /// <param name="ex">異常信息</param>
       public void Debug(string message, Exception ex)
       {
           if (KafkaLogger.IsDebugEnabled) KafkaLogger.Debug(message, ex);
       }
   
       /// <summary>
       ///     Error記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       public void Error(string message)
       {
           if (KafkaLogger.IsErrorEnabled) KafkaLogger.Error(message);
       }
   
       /// <summary>
       ///     Error記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       /// <param name="ex">異常信息</param>
       public void Error(string message, Exception ex)
       {
           if (KafkaLogger.IsErrorEnabled) KafkaLogger.Error(message, ex);
       }
   
       /// <summary>
       ///     Fatal記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       public void Fatal(string message)
       {
           if (KafkaLogger.IsFatalEnabled) KafkaLogger.Fatal(message);
       }
   
       /// <summary>
       ///     Fatal記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       /// <param name="ex">異常信息</param>
       public void Fatal(string message, Exception ex)
       {
           if (KafkaLogger.IsFatalEnabled) KafkaLogger.Fatal(message, ex);
       }
   
       /// <summary>
       ///     Info記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       public void Info(string message)
       {
           if (KafkaLogger.IsInfoEnabled) KafkaLogger.Info(message);
       }
   
       /// <summary>
       ///     Info記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       /// <param name="ex">異常信息</param>
       public void Info(string message, Exception ex)
       {
           if (KafkaLogger.IsInfoEnabled) KafkaLogger.Info(message, ex);
       }
   
       /// <summary>
       ///     Warn記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       public void Warn(string message)
       {
           if (KafkaLogger.IsWarnEnabled) KafkaLogger.Warn(message);
       }
   
       /// <summary>
       ///     Warn記錄
       /// </summary>
       /// <param name="message">日誌信息</param>
       /// <param name="ex">異常信息</param>
       public void Warn(string message, Exception ex)
       {
           if (KafkaLogger.IsWarnEnabled) KafkaLogger.Warn(message, ex);
       }
   
       #endregion Methods
   }

3. 修改Log4Net.Config，定義Kafka的Topic以及Brokers

   <appender name="KafkaAppender" type="MasterChief.DotNet.Core.KafkaLog.KafkaAppender, MasterChief.DotNet.Core.KafkaLog">
           <param name="Topic" value="beats" />
           <param name="Brokers" value="http://localhost:9092" />
           <layout type="log4net.Layout.PatternLayout">
               <conversionPattern value="發生時間：%date %newline事件級別：%-5level %newline事件來源：%logger%newline日誌內容：%message%newline" />
           </layout>
       </appender>

使用

1. 因爲基於上篇說的日誌接口，因此能夠經過Ioc切換，並且不影響在業務代碼調用；
2. 基於業務需求，您能夠同時落地本地日誌，保證網絡抖動或者不正常的時候可以正常記錄日誌；

結語

1. 小弟不才，大佬輕拍；