使用 Queue c# .NET 为消费者和生产者线程创建缓冲区答案

【问题标题】：Creating a buffer for Consumer and Producer threads using Queue c# .NET使用 Queue c# .NET 为消费者和生产者线程创建缓冲区
【发布时间】：2015-11-15 00:18:01
【问题描述】：

我正在编写一个 Windows 服务应用程序，它能够从传感器收集数据，如温度、压力体积等......

读取数据的频率非常高，可能有一百个传感器，接收数据的频率可能是每个传感器每秒一个..

我需要将此数据存储到 oracle 数据库中，原因很明显，我不想以如此高的速度访问数据库。

因此我想创建一个缓冲区。

我的计划是使用标准 .NET Queue 创建一个 Buffer，几个线程将 Enqueue 数据保存到队列中，另一个计时器驱动的线程可以定期写入数据库。

我想知道的是..??这个线程安全吗如果不是这样，创建内存缓冲区的最佳方法是什么

【问题讨论】：

您可能需要阅读 thread-safe collections，特别是 ConcurrentQueue。
几年前我使用ZeroMQ 构建了一个类似的解决方案，它可以解决产品/缺点问题和并发问题。它工作并且仍然工作正常！
添加解决方案后，我阅读了 Micke 的评论，我认为您绝对应该先看看那些 :)

标签： c# .net thread-safety queue

【解决方案1】：

回答您的问题，只要您锁定访问，就可以让多个线程访问一个常规队列。

但对我来说，我没有使用它，而是想使用带锁的队列来保证它们的线程安全。我一直在 c# 中为我的一个程序执行此操作。我只是使用一个常规队列，然后在访问它时放置一个储物柜（入队、出队、计数）。如果你只是锁定访问，它是完全线程安全的。

我的设置来自这里的教程/示例：http://www.albahari.com/threading/part2.aspx#_ProducerConsumerQWaitHandle

我的情况与你的情况有些不同，但非常相似。对我来说，我的数据可以很快进入，如果我不排队，如果多个同时进入，我会丢失数据。然后我有一个线程正在运行，它慢慢地将项目从队列中取出并处理它们。这个切换使用 AutoResetEvent 来保持我的工作线程，直到数据准备好被处理。在您的情况下，您将使用计时器或定期发生的事情。

我复制/粘贴了我的代码并尝试更改名称。希望我没有因为遗漏一些名称更改而完全破坏它，但您应该能够理解要点。

public class MyClass : IDisposable
{
    private Thread sensorProcessingThread = null;
    private Queue<SensorData> sensorQueue = new Queue<SensorData>();
    private readonly object _sensorQueueLocker = new object();
    private EventWaitHandle _whSensorEvent = new AutoResetEvent(false);

    public MyClass () {
        sensorProcessingThread = new Thread(sensorProcessingThread_DoWork);
        sensorProcessingThread.Start();
    }
    public void Dispose()
    {
        // Signal the end by sending 'null'
        EnqueueSensorEvent(null);
        sensorProcessingThread.Join();
        _whSensorEvent.Close();
    }
    // The fast sensor data comes in, locks queue, and then
    // enqueues the data, and releases the EventWaitHandle
    private void EnqueueSensorEvent( SensorData wd )
    {
        lock ( _sensorQueueLocker )
        {
            sensorQueue.Enqueue(wd);
            _whSensorEvent.Set();
        }
    }

    // When asynchronous events come in, I just throw them into queue
    private void OnSensorEvent( object sender, MySensorArgs e )
    {
        EnqueueSensorEvent(new SensorData(sender, e));
    }
    // I have several types of events that can come in,
    // they just get packaged up into the same "SensorData"
    // struct, and I worry about the contents later
    private void FileSystem_Changed( object sender, System.IO.FileSystemEventArgs e )
    {
        EnqueueSensorEvent(new SensorData(sender, e));
    }

    // This is the slower process that waits for new SensorData,
    // and processes it. Note, if it sees 'null' as data,
    // then it knows it should quit the while(true) loop.
    private void sensorProcessingThread_DoWork( object obj )
    {
        while ( true )
        {
            SensorData wd = null;
            lock ( _sensorQueueLocker )
            {
                if ( sensorQueue.Count > 0 )
                {
                    wd = sensorQueue.Dequeue();
                    if ( wd == null )
                    {
                        // Quit the loop, thread finishes
                        return;
                    }
                }
            }
            if ( wd != null )
            {
                try
                {
                    // Call specific handlers for the type of SensorData that was received
                    if ( wd.isSensorDataType1 )
                    {
                        SensorDataType1_handler(wd.sender, wd.SensorDataType1Content);
                    }
                    else
                    {
                        FileSystemChanged_handler(wd.sender, wd.FileSystemChangedContent);
                    }
                }
                catch ( Exception exc )
                {
                    // My sensor processing also has a chance of failing to process completely, so I have a retry
                    // methodology that gives up after 5 attempts
                    if ( wd.NumFailedUpdateAttempts < 5 )
                    {
                        wd.NumFailedUpdateAttempts++;
                        lock ( _sensorQueueLocker )
                        {
                            sensorQueue.Enqueue(wd);
                        }
                    }
                    else
                    {
                        log.Fatal("Can no longer try processing data", exc);
                    }
                }
            }
            else
                _whWatchEvent.WaitOne(); // No more tasks, wait for a signal
        }
    }

您可能会看到来自 Microsoft 的用于 .net 的 Reactive (Rx)。查看：https://msdn.microsoft.com/en-us/data/gg577611.aspx，页面底部是一个 pdf 教程“治愈异步忧郁症”：http://go.microsoft.com/fwlink/?LinkId=208528 这是非常不同的东西，但也许你会看到你喜欢的东西。

【讨论】：