如何有效地异步记录？

Question

我在我的一个项目中使用 Enterprise Library 4 进行日志记录（和其他目的）。我注意到我正在做的日志记录有一些成本，我可以通过在单独的线程上进行日志记录来减轻这些成本。

我现在这样做的方式是创建一个 LogEntry 对象，然后在调用 Logger.Write 的委托上调用 BeginInvoke。

new Action<LogEntry>(Logger.Write).BeginInvoke(le, null, null);

我真正想做的是将日志消息添加到队列中，然后有一个线程将 LogEntry 实例从队列中拉出并执行日志操作。这样做的好处是日志记录不会干扰执行操作，并且不是每个日志记录操作都会导致作业被抛出到线程池中。

如何以线程安全的方式创建一个支持多个写入器和一个读取器的共享队列？非常感谢一些旨在支持多个写入器（不会导致同步/阻塞）和单个读取器的队列实现示例。

也将不胜感激有关替代方法的建议，但我对更改日志记录框架不感兴趣。

Answer 1

这段代码是我不久前写的，请随意使用。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace MediaBrowser.Library.Logging {
    public abstract class ThreadedLogger : LoggerBase {

        Queue<Action> queue = new Queue<Action>();
        AutoResetEvent hasNewItems = new AutoResetEvent(false);
        volatile bool waiting = false;

        public ThreadedLogger() : base() {
            Thread loggingThread = new Thread(new ThreadStart(ProcessQueue));
            loggingThread.IsBackground = true;
            loggingThread.Start();
        }


        void ProcessQueue() {
            while (true) {
                waiting = true;
                hasNewItems.WaitOne(10000,true);
                waiting = false;

                Queue<Action> queueCopy;
                lock (queue) {
                    queueCopy = new Queue<Action>(queue);
                    queue.Clear();
                }

                foreach (var log in queueCopy) {
                    log();
                }
            }
        }

        public override void LogMessage(LogRow row) {
            lock (queue) {
                queue.Enqueue(() => AsyncLogMessage(row));
            }
            hasNewItems.Set();
        }

        protected abstract void AsyncLogMessage(LogRow row);


        public override void Flush() {
            while (!waiting) {
                Thread.Sleep(1);
            }
        }
    }
}

一些优点：

• 它使后台记录器保持活动状态，因此它不需要启动和停止线程。
• 它使用单个线程为队列提供服务，这意味着永远不会出现 100 个线程为队列提供服务的情况。
• 它复制队列以确保在执行日志操作时队列不被阻塞
• 它使用 AutoResetEvent 来确保 bg 线程处于等待状态
• 恕我直言，很容易理解

这是一个略微改进的版本，请记住，我对其进行的测试很少，但它确实解决了一些小问题。

public abstract class ThreadedLogger : IDisposable {

    Queue<Action> queue = new Queue<Action>();
    ManualResetEvent hasNewItems = new ManualResetEvent(false);
    ManualResetEvent terminate = new ManualResetEvent(false);
    ManualResetEvent waiting = new ManualResetEvent(false);

    Thread loggingThread; 

    public ThreadedLogger() {
        loggingThread = new Thread(new ThreadStart(ProcessQueue));
        loggingThread.IsBackground = true;
        // this is performed from a bg thread, to ensure the queue is serviced from a single thread
        loggingThread.Start();
    }


    void ProcessQueue() {
        while (true) {
            waiting.Set();
            int i = ManualResetEvent.WaitAny(new WaitHandle[] { hasNewItems, terminate });
            // terminate was signaled 
            if (i == 1) return; 
            hasNewItems.Reset();
            waiting.Reset();

            Queue<Action> queueCopy;
            lock (queue) {
                queueCopy = new Queue<Action>(queue);
                queue.Clear();
            }

            foreach (var log in queueCopy) {
                log();
            }    
        }
    }

    public void LogMessage(LogRow row) {
        lock (queue) {
            queue.Enqueue(() => AsyncLogMessage(row));
        }
        hasNewItems.Set();
    }

    protected abstract void AsyncLogMessage(LogRow row);


    public void Flush() {
        waiting.WaitOne();
    }


    public void Dispose() {
        terminate.Set();
        loggingThread.Join();
    }
}

相对于原版的优势：

• 它是一次性的，因此您可以摆脱异步记录器
• 改进了刷新语义
• 它会稍微更好地响应突发然后是沉默

Answer 2

是的，您需要一个生产者/消费者队列。我的线程教程中有一个例子——如果你查看我的"deadlocks / monitor methods" 页面，你会在下半部分找到代码。

当然，网上还有很多其他示例 - .NET 4.0 也将在框架中附带一个（比我的功能更全！）。在 .NET 4.0 中，您可能会将 ConcurrentQueue<T> 包装在 BlockingCollection<T> 中。

该页面上的版本是非通用版本（它是在 很久 之前编写的），但您可能希望使其成为通用版本 - 这样做很简单。

你会从每个“正常”线程调用Produce，从一个线程调用Consume，只是循环并记录它消耗的任何内容。让消费者线程成为后台线程可能是最简单的，因此您无需担心应用程序退出时“停止”队列。这确实意味着很可能会丢失最终的日志条目（如果它在应用程序退出时写入它的一半） - 如果您的生产速度超过它可以消耗/记录的速度，甚至更多。

Answer 3

我建议首先测量日志记录对整个系统的实际性能影响（即通过运行分析器），并可选择切换到更快的东西，如log4net（我个人已从EntLib 很久以前的日志记录）。

如果这不起作用，您可以尝试使用 .NET Framework 中的这个简单方法：

ThreadPool.QueueUserWorkItem

将一个方法排队等待执行。该方法在线程池线程可用时执行。

MSDN Details

如果这也不起作用，那么您可以求助于 John Skeet 提供的方法，并自己实际编写异步日志框架。

Answer 4

这是我想出的……另请参阅 Sam Saffron 的回答。这个答案是社区 wiki，以防人们在代码中看到任何问题并想要更新。

/// <summary>
/// A singleton queue that manages writing log entries to the different logging sources (Enterprise Library Logging) off the executing thread.
/// This queue ensures that log entries are written in the order that they were executed and that logging is only utilizing one thread (backgroundworker) at any given time.
/// </summary>
public class AsyncLoggerQueue
{
    //create singleton instance of logger queue
    public static AsyncLoggerQueue Current = new AsyncLoggerQueue();

    private static readonly object logEntryQueueLock = new object();

    private Queue<LogEntry> _LogEntryQueue = new Queue<LogEntry>();
    private BackgroundWorker _Logger = new BackgroundWorker();

    private AsyncLoggerQueue()
    {
        //configure background worker
        _Logger.WorkerSupportsCancellation = false;
        _Logger.DoWork += new DoWorkEventHandler(_Logger_DoWork);
    }

    public void Enqueue(LogEntry le)
    {
        //lock during write
        lock (logEntryQueueLock)
        {
            _LogEntryQueue.Enqueue(le);

            //while locked check to see if the BW is running, if not start it
            if (!_Logger.IsBusy)
                _Logger.RunWorkerAsync();
        }
    }

    private void _Logger_DoWork(object sender, DoWorkEventArgs e)
    {
        while (true)
        {
            LogEntry le = null;

            bool skipEmptyCheck = false;
            lock (logEntryQueueLock)
            {
                if (_LogEntryQueue.Count <= 0) //if queue is empty than BW is done
                    return;
                else if (_LogEntryQueue.Count > 1) //if greater than 1 we can skip checking to see if anything has been enqueued during the logging operation
                    skipEmptyCheck = true;

                //dequeue the LogEntry that will be written to the log
                le = _LogEntryQueue.Dequeue();
            }

            //pass LogEntry to Enterprise Library
            Logger.Write(le);

            if (skipEmptyCheck) //if LogEntryQueue.Count was > 1 before we wrote the last LogEntry we know to continue without double checking
            {
                lock (logEntryQueueLock)
                {
                    if (_LogEntryQueue.Count <= 0) //if queue is still empty than BW is done
                        return;
                }
            }
        }
    }
}

Answer 5

为了回应 Sam Safrons 的帖子，我想打电话给 flush 并确保所有内容都真正完成了编写。就我而言，我正在写入队列线程中的数据库，并且我的所有日​​志事件都在排队，但有时应用程序在所有内容完成写入之前就停止了，这在我的情况下是不可接受的。我更改了您的几块代码，但我想分享的主要内容是刷新：

public static void FlushLogs()
        {   
            bool queueHasValues = true;
            while (queueHasValues)
            {
                //wait for the current iteration to complete
                m_waitingThreadEvent.WaitOne();

                lock (m_loggerQueueSync)
                {
                    queueHasValues = m_loggerQueue.Count > 0;
                }
            }

            //force MEL to flush all its listeners
            foreach (MEL.LogSource logSource in MEL.Logger.Writer.TraceSources.Values)
            {                
                foreach (TraceListener listener in logSource.Listeners)
                {
                    listener.Flush();
                }
            }
        }

我希望这可以减轻一些人的挫败感。在记录大量数据的并行进程中尤其明显。

感谢您分享您的解决方案，它为我指明了一个好的方向！

--约翰尼S

Answer 6

我想说我之前的帖子有点没用。您可以简单地将 AutoFlush 设置为 true，而不必遍历所有侦听器。但是，尝试刷新记录器的并行线程仍然存在疯狂的问题。我必须创建另一个在复制队列和执行 LogEntry 写入期间设置为 true 的布尔值，然后在刷新例程中我必须检查该布尔值以确保队列中没有任何内容并且没有任何内容正在处理在返回之前。

现在并行的多个线程可以命中这个东西，当我调用flush时，我知道它真的被刷新了。

     public static void FlushLogs()
    {
        int queueCount;
        bool isProcessingLogs;
        while (true)
        {
            //wait for the current iteration to complete
            m_waitingThreadEvent.WaitOne();

            //check to see if we are currently processing logs
            lock (m_isProcessingLogsSync)
            {
                isProcessingLogs = m_isProcessingLogs;
            }

            //check to see if more events were added while the logger was processing the last batch
            lock (m_loggerQueueSync)
            {
                queueCount = m_loggerQueue.Count;
            }                

            if (queueCount == 0 && !isProcessingLogs)
                break;

            //since something is in the queue, reset the signal so we will not keep looping

            Thread.Sleep(400);
        }
    }

Answer 7

只是更新：

将企业库 5.0 与 .NET 4.0 一起使用，可以通过以下方式轻松完成：

static public void LogMessageAsync(LogEntry logEntry)
{
    Task.Factory.StartNew(() => LogMessage(logEntry)); 
}

见： http://randypaulo.wordpress.com/2011/07/28/c-enterprise-library-asynchronous-logging/

Answer 8

额外的间接级别可能会有所帮助。

您的第一个异步方法调用可以将消息放入同步队列并设置一个事件——因此锁发生在线程池中，而不是在您的工作线程上——然后另一个线程将消息从队列中拉出引发事件时。

Answer 9

如果你在一个单独的线程上记录一些东西，如果应用程序崩溃，消息可能不会被写入，这使得它相当无用。

原因在于为什么您应该在每次写入后总是刷新。

Answer 10

如果您想到的是一个共享队列，那么我认为您将不得不同步对它的写入、推送和弹出。

但是，我仍然认为值得针对共享队列设计。与日志记录的 IO 相比，并且可能与您的应用程序正在执行的其他工作相比，推送和弹出的短暂阻塞量可能并不重要。