c# - 使用线程和 EventWaitHandle 的生产者/消费者模式

Question

我猜这是一种代码审查，但这是我对生产者/消费者模式的实现。我想知道的是，在 ReceivingThread() 或 SendingThread() 方法中的 while 循环是否会停止执行。请注意，EnqueueSend(DataSendEnqeueInfo info) 是从多个不同的线程调用的，我可能无法在此处使用任务，因为我肯定必须在单独的线程中使用命令。

private Thread mReceivingThread;
private Thread mSendingThread;
private Queue<DataRecievedEnqeueInfo> mReceivingThreadQueue;
private Queue<DataSendEnqeueInfo> mSendingThreadQueue;
private readonly object mReceivingQueueLock = new object();
private readonly object mSendingQueueLock = new object();
private bool mIsRunning;
EventWaitHandle mRcWaitHandle;
EventWaitHandle mSeWaitHandle;

private void ReceivingThread()
{
    while (mIsRunning)
    {
        mRcWaitHandle.WaitOne();
        DataRecievedEnqeueInfo item = null;
        while (mReceivingThreadQueue.Count > 0)
        {
            lock (mReceivingQueueLock)
            {
                item = mReceivingThreadQueue.Dequeue();
            }
            ProcessReceivingItem(item);
        }
        mRcWaitHandle.Reset();
    }
}

private void SendingThread()
{
    while (mIsRunning)
    {
        mSeWaitHandle.WaitOne();
        while (mSendingThreadQueue.Count > 0)
        {
            DataSendEnqeueInfo item = null;
            lock (mSendingQueueLock)
            {
                item = mSendingThreadQueue.Dequeue();
            }
            ProcessSendingItem(item);
        }
        mSeWaitHandle.Reset();
    }
}

internal void EnqueueRecevingData(DataRecievedEnqeueInfo info)
{
    lock (mReceivingQueueLock)
    {
        mReceivingThreadQueue.Enqueue(info);
        mRcWaitHandle.Set();
    }
}

public void EnqueueSend(DataSendEnqeueInfo info)
{
     lock (mSendingQueueLock)
    {
        mSendingThreadQueue.Enqueue(info);
        mSeWaitHandle.Set();
    }
}

P.S 这里的想法是，当队列为空时，我使用 WaitHandle 将线程置于休眠状态，并在新项目入队时发出信号让它们启动。

更新我要离开这个https://blogs.msdn.microsoft.com/benwilli/2015/09/10/tasks-are-still-not-threads-and-async-is-not-parallel/ ，对于可能尝试使用 TPL 或任务实现生产者/消费者模式的人。

Answer 1

使用 BlockingCollection代替 Queue、EventWaitHandle 和锁定对象:

public class DataInfo { }

private Thread mReceivingThread;
private Thread mSendingThread;

private BlockingCollection<DataInfo> queue;

private CancellationTokenSource receivingCts = new CancellationTokenSource();

private void ReceivingThread()
{
    try
    {
        while (!receivingCts.IsCancellationRequested)
        {
            // This will block until an item is added to the queue or the cancellation token is cancelled
            DataInfo item = queue.Take(receivingCts.Token);

            ProcessReceivingItem(item);
        }
    }
    catch (OperationCanceledException)
    {

    }
}

internal void EnqueueRecevingData(DataInfo info)
{
    // When a new item is produced, just add it to the queue
    queue.Add(info);
}

// To cancel the receiving thread, cancel the token
private void CancelReceivingThread()
{
    receivingCts.Cancel();
}