如何使用 BlockingCollection<> 解决生产者/消费者竞争条件

Question

我正在实现一个将记录写入数据库的记录器。为了防止数据库写入阻塞调用记录器的代码，我已将数据库访问移至单独的线程，使用基于BlockingCollection<string> 的生产者/消费者模型实现。

这是简化的实现：

abstract class DbLogger : TraceListener
{
    private readonly BlockingCollection<string> _buffer;
    private readonly Task _writerTask;

    DbLogger() 
    {
        this._buffer = new BlockingCollection<string>(new ConcurrentQueue<string>(), 1000);
        this._writerTask = Task.Factory.StartNew(this.ProcessBuffer, TaskCreationOptions.LongRunning);
    }

    // Enqueue the msg.
    public void LogMessage(string msg) { this._buffer.Add(msg); }

    private void ProcessBuffer()
    {
        foreach (string msg in this._buffer.GetConsumingEnumerable())
        {
            this.WriteToDb(msg);
        }
    }

    protected abstract void WriteToDb(string msg);

    protected override void Dispose(bool disposing) 
    { 
        if (disposing) 
        {
            // Signal to the blocking collection that the enumerator is done.
            this._buffer.CompleteAdding();

            // Wait for any in-progress writes to finish.
            this._writerTask.Wait(timeout);

            this._buffer.Dispose(); 
        }
        base.Dispose(disposing); 
    }
}

现在，当我的应用程序关闭时，我需要确保在数据库连接断开之前刷新缓冲区。否则，WriteToDb 会抛出异常。

所以，这是我幼稚的 Flush 实现：

public void Flush()
{
    // Sleep until the buffer is empty.
    while(this._buffer.Count > 0)
    {
        Thread.Sleep(50);
    }
}

此实现的问题在于以下事件序列：

1. 缓冲区中有一个条目。
2. 在日志记录线程中，MoveNext() 在枚举器上被调用，因此我们现在位于 ProcessBuffer 的 foreach 循环的主体中。
3. Flush() 由主线程调用。它看到集合是空的，所以立即返回。
4. 主线程关闭数据库连接。
5. 回到日志线程，foreach 循环的主体开始执行。 WriteToDb 被调用，但由于数据库连接已关闭而失败。

所以，我的下一个尝试是添加一些标志，如下所示：

private volatile bool _isWritingBuffer = false;
private void ProcessBuffer()
{
    foreach (string msg in this._buffer.GetConsumingEnumerable())
    {
        lock (something) this._isWritingBuffer = true;
        this.WriteToDb(msg);
        lock (something) this._isWritingBuffer = false;
    }
}

public void Flush()
{
    // Sleep until the buffer is empty.
    bool isWritingBuffer;
    lock(something) isWritingBuffer = this._isWritingBuffer;
    while(this._buffer.Count > 0 || isWritingBuffer)
    {
        Thread.Sleep(50);
    }
}

但是，仍然存在竞争条件，因为整个 Flush() 方法可以在集合为空之后但在 _isWritingBuffer 设置为 true 之前执行。

如何修复我的 Flush 实现以避免这种竞争条件？

注意：由于各种原因，我必须从头开始编写记录器，所以请不要回答建议我使用一些现有的记录框架。

Answer 1

首先永远不要锁定公共对象，尤其是this。

此外永远不要使用裸布尔值进行同步：如果您想了解可能出现的问题，请参阅我的博客：Synchronization, memory visibility and leaky abstractions :)

关于问题本身，我一定遗漏了一些东西，但为什么需要这样的Flush 方法？

确实，当您完成日志记录后，您将通过从主线程调用其Dispose 方法来处理记录器。

并且您已经以等待“写入数据库”任务的方式实现它。

如果我错了，你真的需要与另一个原语同步，那么你应该使用一个事件：

在DbLogger：

public ManualResetEvent finalizing { get; set; }

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

在某个地方，例如在 ProcessBuffer 中，您在完成写入 DB 时会通知您：

finalizing.Set();