并发队列 C# 中的线程安全

Question

我有一个MessagesManager 线程，不同的线程可以向它发送消息，然后这个MessagesManager 线程负责在SendMessageToTcpIP() 中发布这些消息（MessagesManager 线程的起点）。

class MessagesManager : IMessageNotifier
{
    //private
    private readonly AutoResetEvent _waitTillMessageQueueEmptyARE = new AutoResetEvent(false);
    private ConcurrentQueue<string> MessagesQueue = new ConcurrentQueue<string>(); 

    public void PublishMessage(string Message)
    {
        MessagesQueue.Enqueue(Message);
        _waitTillMessageQueueEmptyARE.Set();
    }

    public void SendMessageToTcpIP()
    {
        //keep waiting till a new message comes
        while (MessagesQueue.Count() == 0)
        {
            _waitTillMessageQueueEmptyARE.WaitOne();
        }

        //Copy the Concurrent Queue into a local queue - keep dequeuing the item once it is inserts into the local Queue
        Queue<string> localMessagesQueue = new Queue<string>();

        while (!MessagesQueue.IsEmpty)
        {
            string message;
            bool isRemoved = MessagesQueue.TryDequeue(out message);
            if (isRemoved)
                localMessagesQueue.Enqueue(message);
        }

        //Use the Local Queue for further processing
        while (localMessagesQueue.Count() != 0)
        {
            TcpIpMessageSenderClient.ConnectAndSendMessage(localMessagesQueue.Dequeue().PadRight(80, ' '));
            Thread.Sleep(2000);
        }
    }
}

不同的线程 (3-4) 通过调用 PublishMessage(string Message) 发送它们的消息（使用相同的对象到 MessageManager）。消息到达后，我将该消息推送到并发队列，并通过设置_waitTillMessageQueueEmptyARE.Set(); 通知SendMessageToTcpIP()。 在SendMessageToTcpIP()里面，我是从本地队列里面的并发队列中复制消息，然后一一发布。

问题：以这种方式进行入队和出队是否线程安全？会不会有什么奇怪的效果？

Answer 1

虽然这可能是线程安全的，但 .NET 中有内置类可帮助实现“多个发布者一个消费者”模式，例如 BlockingCollection。你可以像这样重写你的类：

class MessagesManager : IDisposable {
    // note that your ConcurrentQueue is still in play, passed to constructor
    private readonly BlockingCollection<string> MessagesQueue = new BlockingCollection<string>(new ConcurrentQueue<string>());

    public MessagesManager() {
        // start consumer thread here
        new Thread(SendLoop) {
            IsBackground = true
        }.Start();
    }

    public void PublishMessage(string Message) {
        // no need to notify here, will be done for you
        MessagesQueue.Add(Message);
    }

    private void SendLoop() {
        // this blocks until new items are available
        foreach (var item in MessagesQueue.GetConsumingEnumerable()) {
            // ensure that you handle exceptions here, or whole thing will break on exception
            TcpIpMessageSenderClient.ConnectAndSendMessage(item.PadRight(80, ' '));
            Thread.Sleep(2000); // only if you are sure this is required 
        }
    }

    public void Dispose() {            
        // this will "complete" GetConsumingEnumerable, so your thread will complete
        MessagesQueue.CompleteAdding();
        MessagesQueue.Dispose();
    }
}

Answer 2

.NET 已经提供了ActionBlock< T>，它允许将消息发布到缓冲区并异步处理它们。默认情况下，一次只处理一条消息。

你的代码可以改写为：

//In an initialization function
ActionBlock<string> _hmiAgent=new ActionBlock<string>(async msg=>{
        TcpIpMessageSenderClient.ConnectAndSendMessage(msg.PadRight(80, ' '));
        await Task.Delay(2000);
);

//In some other thread ...
foreach ( ....)
{
    _hmiAgent.Post(someMessage);
}

// When the application closes

_hmiAgent.Complete();
await _hmiAgent.Completion;

ActionBlock 提供了许多好处 - 您可以指定它可以在缓冲区中接受的项目数量的限制，并指定可以并行处理多条消息。您还可以在处理管道中组合多个块。在桌面应用程序中，可以将消息发布到管道以响应事件，由单独的块处理，并将结果发布到更新 UI 的最终块。

例如，填充可以由中介TransformBlock< TIn,TOut> 执行。这种转换是微不足道的，使用块的成本比方法大，但这只是一个说明：

//In an initialization function
TransformBlock<string> _hmiAgent=new TransformBlock<string,string>(
    msg=>msg.PadRight(80, ' '));

ActionBlock<string> _tcpBlock=new ActionBlock<string>(async msg=>{
        TcpIpMessageSenderClient.ConnectAndSendMessage());
        await Task.Delay(2000);
);

var linkOptions=new DataflowLinkOptions{PropagateCompletion = true};
_hmiAgent.LinkTo(_tcpBlock);

发布代码根本没有改变

    _hmiAgent.Post(someMessage);

当应用程序终止时，我们需要等待_tcpBlock完成：

    _hmiAgent.Complete();
    await _tcpBlock.Completion;

Visual Studio 2015+ 本身使用 TPL Dataflow 处理此类场景

Bar Arnon 在TPL Dataflow Is The Best Library You're Not Using 中提供了一个更好的示例，它展示了如何在块中同时使用同步和异步方法。

Answer 3

代码是线程安全的，因为ConcurrentQueue 和AutoResetEvent 都是线程安全的。无论如何，您的字符串都被读取并且永远不会被写入，因此此代码是线程安全的。

但是，您必须确保在某种循环中调用 SendMessageToTcpIP。
否则，您将面临危险的竞争条件 - 一些消息可能会丢失：

while (!MessagesQueue.IsEmpty)
        {
            string message;
            bool isRemoved = MessagesQueue.TryDequeue(out message);
            if (isRemoved)
                localMessagesQueue.Enqueue(message);
        }

        //<<--- what happens if another thread enqueues a message here?

        while (localMessagesQueue.Count() != 0)
        {
            TcpIpMessageSenderClient.ConnectAndSendMessage(localMessagesQueue.Dequeue().PadRight(80, ' '));
            Thread.Sleep(2000);
        }

除此之外，AutoResetEvent 是非常重的物体。它使用内核对象来同步线程。每个调用都是一个系统调用，可能代价高昂。考虑使用用户模式同步对象（.net 不提供某种条件变量吗？）

Answer 4

这是我将如何实现此功能的重构代码 sn-p：

class MessagesManager {
    private readonly AutoResetEvent messagesAvailableSignal = new AutoResetEvent(false);
    private readonly ConcurrentQueue<string> messageQueue = new ConcurrentQueue<string>();

    public void PublishMessage(string Message) {
        messageQueue.Enqueue(Message);
        messagesAvailableSignal.Set();
    }

    public void SendMessageToTcpIP() {
        while (true) {
            messagesAvailableSignal.WaitOne();
            while (!messageQueue.IsEmpty) {
                string message;
                if (messageQueue.TryDequeue(out message)) {
                    TcpIpMessageSenderClient.ConnectAndSendMessage(message.PadRight(80, ' '));
                }
            }
        }
    }
}

这里需要注意的地方：

1. 这会完全排空队列：如果至少有一条消息，它将处理所有消息
2. 2000ms 线程休眠被移除