我在 2 个线程之间使用 ConcurrentQueue 可以吗？

Question

我可以在 2 个线程之间使用 ConcurrentQueue 吗？我想检查我不需要明确地“锁定”任何地方。特别是看看我在 COMMENTS 中的行，我会在这里丢包吗...

public class PacketCapturer
{

private static ConcurrentQueue<Packet> _packetQueue = new ConcurrentQueue<Packet>();

public PacketCapturer(IPHostEntry proxyDns, ref BackgroundWorker bw)
{
    // start the background thread
    var backgroundThread = new System.Threading.Thread(BackgroundThread);
    backgroundThread.Start();

    // Start Packet Capture with callback via PacketCapturerCallback
}

private void PacketCapturerCallback(Packet packet)
{
    _packetQueue.Enqueue(packet);
}

private static void BackgroundThread()
{
    while (!BackgroundThreadStop)
    {
        if (_packetQueue.Count == 0) Thread.Sleep(250);
        else 
        {
            ConcurrentQueue<Packet> ourQueue;  
            ourQueue = _packetQueue;   // COULD I DROP A PACKET BETWEEN HERE
            _packetQueue = new ConcurrentQueue<Packet>();      // AND HERE???

            Console.WriteLine("BackgroundThread: ourQueue.Count is {0}", ourQueue.Count);
        }
    }
}

Answer 1

不，不行。首先，如果您在并发线程中像这样更改引用，则 _packetQueue 必须 标记为 volatile 以防止编译器和代码生成优化永远不会看到更改。 _packetQueue 的 cahning 通常应作为 Interlocked.CompareExchange 发生，但这对您的使用不太重要。

但更令人担忧的是在后台线程中像这样更改 packetQueue 实例的模式。这样做的目的是什么？它有一种可怕的代码气味......

更新

我通常做的是：

生产者线程：

Producer () {
...
lock(_sharedQueue) {
  _sharedQueue.Enqueue(something);
}
...
}

消费者线程：

consumer (...) {
...
var Something[] toProcess = null;
lock(_sharedQueue)
{
  toProcess = _sharedQueue.Toarray();
   _sharedQueue.Clear();
}
// Process here the toProcess array
...
}

这对于我曾经有过的每一次使用都足够好。处理不会在锁定下发生，因此锁定是最小的。不需要花哨的 ConcurrentQueue，一个普通的旧 .Net 2.0 集合就足够了。通常我会使用专用的锁定对象，而不是锁定实际的队列实例。

Answer 2

编辑：我认为 Mark/Gabe 是正确的，您不会丢失任何数据包。我会把剩下的留作参考，以防其他人对此发表意见。

---

很简单，是的。您可能会在那里丢失一个或多个数据包。您可能想查看 ConcurrentQueue 提供哪些方法来获取/删除其中的一部分，因为这看起来像您想要做的。

为什么不 TryDequeue 直到它返回 false：

    else 
    {
        Queue<Packet> ourQueue = new Queue<Packet>();  //this doesn't need to be concurrent unless you want it to be for some other reason.
        Packet p;
        while(_packetQueue.TryDequeue(out p))
        {
            ourQueue.Enqueue(p);
        }

        Console.WriteLine("BackgroundThread: ourQueue.Count is {0}", ourQueue.Count);
    }

Answer 3

我认为你不能丢弃一个数据包，因为显然任何线程都会取消对旧的或新的的引用。边缘情况是，您在ourQueue 收到更多数据 你认为你已经交换了它们，或者你也可能最终没有注意到Sleep 循环中的参考变化（但是: 有2个线程，唯一改变引用的线程是这个线程，所以不一定有问题。

TBH，如果您在这里有 2 个线程，为什么不直接使用 lock 或使用 ReaderWriterLock？弄清楚什么时候会发生什么会简单得多……