使用线程和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();
    }
}

这里的想法是，当队列为空时，使用WaitHandle使线程休眠，并在入队新项时发出信号以启动它们。

更新 我只是留下了https://blogs.msdn.microsoft.com/benwilli/2015/09/10/tasks-are-still-not-threads-and-async-is-not-parallel/，供那些试图使用TPL或任务实现生产者/消费者模式的人参考。

- Sushant Poojary

---

你最好使用一个 BlockingCollection，它可以为你处理所有的同步逻辑。 - Ilian

阻塞集合是一个不错的选择，但你也可以尝试一下TPL Dataflow。如果需要，我可以提供一些示例逻辑。 - VMAtm

@VMAtm 我一直想弄懂TPL数据流。如果可以的话，您能否提供一些示例逻辑给我呢？当然，只要不太麻烦。 - Sushant Poojary

@SushantPoojary 我已经添加了答案并附上了一些示例。 - VMAtm

嗨，Sushant Poojary，请查看此代码。 - sɐunıɔןɐqɐp

Answer 1

使用 BlockingCollection 代替 Queue、EventWaitHandle 和 lock 对象：

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();
}

- Serge

---

您有一个很好的BlockingCollection示例，您能进一步说明如何启动此单独的线程吗？ 您会如何从另一个线程向此集合添加项？ - Ozkan

Answer 2

对于简单的生产者-消费者问题，我建议直接使用BlockingCollection。不需要手动编写同步逻辑，如果队列中没有元素，消费线程也会阻塞。

如果使用此类，您的代码可能如下所示：

private BlockingCollection<DataRecievedEnqeueInfo> mReceivingThreadQueue = new BlockingCollection<DataRecievedEnqeueInfo>();
private BlockingCollection<DataSendEnqeueInfo> mSendingThreadQueue = new BlockingCollection<DataSendEnqeueInfo>();

public void Stop()
{
    // No need for mIsRunning. Makes the enumerables in the GetConsumingEnumerable() calls
    // below to complete.
    mReceivingThreadQueue.CompleteAdding();
    mSendingThreadQueue.CompleteAdding();
}

private void ReceivingThread()
{
    foreach (DataRecievedEnqeueInfo item in mReceivingThreadQueue.GetConsumingEnumerable())
    {
        ProcessReceivingItem(item);
    }
}

private void SendingThread()
{
    foreach (DataSendEnqeueInfo item in mSendingThreadQueue.GetConsumingEnumerable())
    {
        ProcessSendingItem(item);
    }
}

internal void EnqueueRecevingData(DataRecievedEnqeueInfo info)
{
    // You can also use TryAdd() if there is a possibility that you
    // can add items after you have stopped. Otherwise, this can throw an
    // an exception after CompleteAdding() has been called.
    mReceivingThreadQueue.Add(info);
}

public void EnqueueSend(DataSendEnqeueInfo info)
{
    mSendingThreadQueue.Add(info);
}

- Ilian

---

如果我没记错的话，BlockingCollection.GetConsumingEnumerable() 返回的集合顺序不一定与入队顺序相同，这对我来说是个问题。另外，当集合为空时，我希望线程休眠，所以我正在使用 Waithandles，那么这也会被内部处理吗？还要进行基准测试，但在此之前，您是否知道 BlockingCollections 的性能表现更好还是更差？ - Sushant Poojary

BlockingCollection将利用您提供的任何IProducerConsumerCollection<T>。默认情况下，它将使用ConcurrentQueue<T>，因此它将按照入队顺序进行处理。与您编写解决方案的方式相同。性能方面？只有您可以对其进行分析。但我愿意打赌，与您（或我）手动实现的任何内容相比，它至少会更快。 - Ilian

哇！BlockingCollection.GetConsumingEnumerable() 自动阻塞线程...我不知道为什么总是害怕在并发和/或线程方面使用较新的 API。-_-。感谢您的时间，我学到了新东西。 - Sushant Poojary

Answer 3

根据评论建议，您也可以尝试使用TPL Dataflow块。

就我所见，您有两个类似的管道，一个用于接收，另一个用于发送，因此我假设您的类层次结构如下：

class EnqueueInfo { }
class DataRecievedEnqeueInfo : EnqueueInfo { }
class DataSendEnqeueInfo : EnqueueInfo { }

我们可以组装一个抽象类，将创建管道的逻辑封装起来，并提供处理项的接口，如下所示：

abstract class EnqueueInfoProcessor<T>
    where T : EnqueueInfo
{
    // here we will store all the messages received before the handling
    private readonly BufferBlock<T> _buffer;
    // simple action block for actual handling the items
    private ActionBlock<T> _action;

    // cancellation token to cancel the pipeline
    public EnqueueInfoProcessor(CancellationToken token)
    {
        _buffer = new BufferBlock<T>(new DataflowBlockOptions { CancellationToken = token });
        _action = new ActionBlock<T>(item => ProcessItem(item), new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = Environment.ProcessorCount,
            CancellationToken = token
        });

        // we are linking two blocks so all the items from buffer
        // will flow down to action block in order they've been received
        _buffer.LinkTo(_action, new DataflowLinkOptions { PropagateCompletion = true });
    }

    public void PostItem(T item)
    {
        // synchronously wait for posting to complete
        _buffer.Post(item);
    }

    public async Task SendItemAsync(T item)
    {
        // asynchronously wait for message to be posted
        await _buffer.SendAsync(item);
    }

    // abstract method to implement
    protected abstract void ProcessItem(T item);
}

请注意，您还可以使用 Encapsulate<TInput, TOutput> 方法封装两个块之间的链接，但在这种情况下，如果您正在使用它，则必须正确处理缓冲块的 Completion。

之后，我们只需要实现接收和发送处理逻辑的两个方法即可：

public class SendEnqueueInfoProcessor : EnqueueInfoProcessor<DataSendEnqeueInfo>
{
    SendEnqueueInfoProcessor(CancellationToken token)
        : base(token)
    {

    }
    protected override void ProcessItem(DataSendEnqeueInfo item)
    {
        // send logic here
    }
}

public class RecievedEnqueueInfoProcessor : EnqueueInfoProcessor<DataRecievedEnqeueInfo>
{
    RecievedEnqueueInfoProcessor(CancellationToken token)
        : base(token)
    {

    }
    protected override void ProcessItem(DataRecievedEnqeueInfo item)
    {
        // recieve logic here
    }
}

如果您的消息流是关于ReceiveInfo消息变成SendInfo，那么您还可以使用TransformBlock<DataRecievedEnqeueInfo, DataSendEnqeueInfo>创建更复杂的管道。

- VMAtm

---

哇！非常感谢您提供如此详细的示例。我已经在 MSDN 上阅读了几天有关 TransformBlock 的内容，看起来非常有趣。我将实现此代码并查看其工作原理。 - Sushant Poojary