如何提高同时访问项目集合的两个线程的性能

Question

我正在 .net 项目 3.5 上实施生产者/消费者方法

只有一个生产者和一个消费者在各自的线程上运行

CheckOrderToProcess 方法检查符合特定条件的表并将它们添加到列表中（生产者）

bgOrdenes_DoWork 方法获取列表中的每个项目并执行一些逻辑（消费者）

我想避免锁定整个列表lstOrderToProcessto 以提高性能，我尝试使用ConcurrentQueue，但我无法升级到.net 4.0，因为这将用于不能升级到4.0

如何更改此实现以提高性能？我不必是一个列表，只要它可以在线程之间共享，我可以在最后添加并获取第一个元素（队列）

public class DatabaseCache : ICacheDB
{
    private static List<NotificacionOrdenes> lstOrderToProcess;
    private static Object lockOrders;
    private static BackgroundWorker bgOrdenes;
    private static bool pendingOrders = false;
    private static Timer timerCheck;

    public DatabaseCache(string eventSourceName, bool monitorearOrderNotifications)
    {
        bgOrdenes = new BackgroundWorker();
        bgOrdenes.DoWork += new DoWorkEventHandler(bgOrdenes_DoWork);

        lstOrderToProcess = new List<NotificacionOrdenes>();
        lockOrders = new Object();

        CheckOrderToProcess();

        if (!bgOrdenes.IsBusy)
        {
            bgOrdenes.RunWorkerAsync();
        }

        //execute CheckOrderToProcess periodically
        timerCheck = new Timer(2000);
        timerCheck.Elapsed += new ElapsedEventHandler(timerCheck_Elapsed);
        timerCheck.Enabled = true;
        timerCheck.AutoReset = true;

    }
    void timerCheck_Elapsed(object sender, ElapsedEventArgs e)
    {
        CheckOrderToProcess();
    }

    private void CheckOrderToProcess()
    {
        DataSet ds;
        NotificacionOrdenes notif;

        ds = Data.GetOrderNotifications_ToProcess();

        //if there is new info to process
        if ((ds != null) && (ds.Tables[0].Rows.Count != 0))
        {

            foreach (DataRow row in ds.Tables[0].Rows)
            {
                notif = new NotificacionOrdenes();

                //fill NOTIF with info of each row

                lock (lockOrders)
                {
                    lstOrderToProcess.Add(notif);
                }

            }

            pendingOrders = true;
        }
    }

    void bgOrdenes_DoWork(object sender, DoWorkEventArgs e)
    {
        NotificacionOrdenes notif;

        while (true)
        {
            if (pendingOrders)
            {
                lock (lockOrders)
                {
                    notif = lstOrderToProcess[0];
                    lstOrderToProcess.RemoveAt(0);

                    //check to see if there is any pending order
                    pendingOrders = lstOrderToProcess.Any();
                }

                //Execute rest of the logic
            }
        }
    }
}

Answer 1

如果我会在你的情况下做生产者/消费者，那么我也会从List<> 开始。

但正如我现在所看到的，这不是最好的情况，因为从列表中删除项目会导致索引被更改（这需要锁定整个列表）。

也许你可以使用数组来代替？像这样的：

NotificacionOrdenes[] todo = new NotificacionOrdenes[1000];

// producer
// find empty slot (null) in todo somehow
todo[empty_slot] = new NotificacionOrdenes();
...

// consumer
// find non-empty slot somehow
var notif = todo[non_empty_slot]
todo[non_empty_slot] = null;
..

如您所见，不需要锁（检查 null 并设置为 null 是安全的），但是您必须处理没有空槽、数组太小或消费者太慢的情况。

Answer 2

没有可以想象的理由来更改此代码 sn-p 中的锁定。然而，它确实存在相当严重的设计缺陷：

• DoWork 内部的等待循环是一个热等待循环。它燃烧 100% 核心，99.99999% 的时间什么都不做。这通常对您运行此代码的机器非常不友好。它还会导致您的程序对添加的项目无响应，即使您在尝试检测时消耗了大量的 CPU 周期。在你烧掉你的量子后，操作系统的线程调度程序会让你在狗屋里呆一段时间。

• pendingOrders 变量用作同步对象，但只是一个简单的bool 变量。当你这样做时，很多事情都会出错。一方面，当您运行代码的 Release 版本时，您的代码很可能永远不会看到该变量被设置为 true。 32 位代码中的一个问题，这样的变量必须声明为volatile。这也是低效的，线程可能需要一段时间才能观察到分配的值。

• 使用 lstOrderToProcess.Any() 效率低下。当您实际上将清空整个列表时，仅删除索引 0 处的元素是没有意义的。

• 消费者在 BackgroundWorker 上运行。它使用线程池线程来实现工作者。 TP 线程通常不应运行超过约半秒。但是，您的线程永远运行，严重阻碍了线程池调度程序执行待处理的 tp 线程请求的工作。这会对整个应用的响应能力产生负面影响。

通过使用常规线程来运行消费者来获得成功。并为列表使用更好的设计，您需要一个阻塞队列。您可以使用线程大师的代码获得在旧 .NET 版本上运行的一个，Joe Duffy 的sample implementation 解决了您的热等待循环问题。我将在此处重新发布，经过调整以不等待消费者：

public class BlockingQueue<T> { 
    private Queue<Cell<T>> m_queue = new Queue<Cell<T>>(); 
    public void Enqueue(T obj) { 
        Cell<T> c = new Cell<T>(obj);
        lock (m_queue) {
            m_queue.Enqueue(c); 
            Monitor.Pulse(m_queue); 
        }
    } 
    public T Dequeue() { 
        Cell<T> c; 
        lock (m_queue) { 
            while (m_queue.Count == 0) Monitor.Wait(m_queue); 
            c = m_queue.Dequeue();
        } 
        return c.m_obj; 
    } 
}

class Cell<T> { 
    internal T m_obj; 
    internal Cell(T obj) { 
        m_obj = obj;
    } 
}