断锁策略——分析修正

Question

我问这个主要是为了进行健全性检查：在 C# (8.0) 应用程序中，我有这段代码，它虚假地失败，出现来自 Monitor.pulse() 的“对象未同步”异常（我已经为了清楚起见，省略了不相关的代码）：

// vanilla multiple-producer single-consumer queue stuff:

private Queue<Message> messages = new Queue<Message>();

private void ConsumerThread () {
    Queue<Message> myMessages = new Queue<Message>();
    while (...) {
        lock (messages) {
            // wait
            while (messages.Count == 0)
                Monitor.Wait(messages);
            // swap
            (messages, myMessages) = (myMessages, messages);
        }
        // process
        while (myMessages.Count > 0)
            DoStuff(myMessages.Dequeue());
    }
}

public void EnqueueMessage (...) {
    Message message = new Message(...);
    lock (messages) {
        messages.Enqueue(message);
        Monitor.Pulse(messages);
    }
}

我对 C# 还很陌生，而且我在写它的时候也很紧张。现在我正在审查该代码以修复异常，我立即对我在消费者的锁内重新分配messages 感到惊讶。

我环顾四周，发现Is it bad to overwrite a lock object if it is the last statement in the lock?，这验证了我的眉毛。

---

不过，我还是没有太大的信心（经验不足+压力大），所以，只是为了确认一下：下面分析为什么会这样坏的正确吗？

如果发生以下情况，按以下顺序：

1. 东西恰好在队列中。
2. 消费者线程锁定messages（并将跳过等待循环）。
3. EnqueueMessage 尝试锁定 messages，等待锁定。
4. 消费者线程交换messages和myMessages，释放锁。
5. EnqueueMessage 锁定。
6. EnqueueMessage 将 item 添加到 messages 并调用 Monitor.pulse(messages) 除了 messages 不是它在步骤 (3) 中锁定的对象，因为它是在 (4) 中从我们下面换出的。可能的后果包括：
   • 在非锁定对象（以前是 myMessages）上调用 Monitor.Pulse - 因此出现上述异常。
   • 排队到错误的队列及其后果。
   • 如果消费者线程设法完成另一个完整的循环周期，而 EnqueueMessage 仍位于其 lock{} 中的某个位置，那就更奇怪了。

对吗？我很确定这是对的，感觉很基础，但我只是想确认一下，因为我现在完全筋疲力尽了。

---

那么，无论这是否正确：以下建议的修复有意义吗？

在我看来，解决方法非常简单：不要使用 messages 作为监控对象，只需使用一些不会更改的专用虚拟对象即可：

private readonly object messagesLock = new object();
private Queue<Message> messages = new Queue<Message>();

private void ConsumerThread () {
    Queue<Message> myMessages = new Queue<Message>();
    while (...) {
        lock (messagesLock) {
            while (messages.Count == 0) 
                Monitor.Wait(messagesLock);
            (messages, myMessages) = (myMessages, messages);
        }
    }
    ...
}

public void EnqueueMessage (...) {
    ...;
    lock (messagesLock) {
        messages.Enqueue(...);
        Monitor.Pulse(messagesLock);
    }
}

其目的是避免因在陌生位置换出锁定对象而引起的任何问题。

那应该可以……对吧？

Answer 1

大概 16 年前的 .NET 2 以来，没有人在多线程中使用 Queue（如果我的日期有误，请纠正我）。

并发集合很简单。

BlockingColleciton<Message> myMessages = new BlockingColleciton<Message>();

private void ConsumerThread () {
    while (...) 
    {
        var message = myMessages.Take();
    }
    ...
}

public void EnqueueMessage (Message msg) {
    ...;
    myMessages.Add(msg);
}