c# - C#中的多线程应用程序

Question

我已经创建了一个用于优化 pdf 文件的 .net 应用程序。实际上我必须优化很多文件并且我调用了这样一个线程:

CheckForIllegalCrossThreadCalls = false;
thOptimize = new Thread(csCommon.pdfFilesCompressAndMove);
thOptimize.Start();

我也找到了没有。使用这个的处理器和内核:

int processors=Environment.ProcessorCount
int coreCount = 0;
foreach (var item in new System.Management.ManagementObjectSearcher("Select * from  Win32_Processor").Get())
{
coreCount += int.Parse(item["NumberOfCores"].ToString());
}

我在我的机器中发现了 4 个处理器和 2 个内核。

现在我的问题是我想对所有处理器使用函数pdfFilesCompressAndMove 即我想同时优化多个文件。换句话说，我想让所有处理器保持忙碌在优化中。

请指导我怎么可能？

Answer 1

你想要的是生产者/消费者队列。

这里发生的是生产者创建工作项供消费者处理。当生产者可以比消费者处理它的速度更快地为消费者创建工作时，这很有效。然后，您有一个或多个消费者处理这个工作队列。

这是我用于此类事情的生产者消费者类:

public class ProducerConsumer<T>:IDisposable 
    {
        private  int _consumerThreads;
        private readonly Queue<T> _queue = new Queue<T>();
        private readonly object _queueLocker = new object();
        private readonly AutoResetEvent _queueWaitHandle = new AutoResetEvent(false);
        private readonly Action<T> _consumerAction;
        private readonly log4net.ILog _log4NetLogger = log4net.LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);
        private bool _isProcessing = true;

        public ProducerConsumer(Action<T> consumerAction,int consumerThreads,bool isStarted)
        {
            _consumerThreads = consumerThreads;

            if (consumerAction == null)
            {
                throw new ArgumentNullException("consumerAction");
            }
            _consumerAction = consumerAction;
            if (isStarted)
                Start();
            //just in case the config item is missing or is set to 0.  We don't want to have the queue build up
        }

        public ProducerConsumer(Action<T> consumerAction, int consumerThreads):this(consumerAction,consumerThreads,true)
        {


        }
        public void Dispose()
        {
            _isProcessing = false;
            lock(_queueLocker)
            {
                _queue.Clear();
            }
        }
        public void Start()
        {
            if (_consumerThreads == 0)
                _consumerThreads = 2;

            for (var loop = 0; loop < _consumerThreads; loop++)
                ThreadPool.QueueUserWorkItem(ConsumeItems);
        }

        public void Enqueue(T item)
        {
            lock (_queueLocker)
            {
                _queue.Enqueue(item);
                // After enqueuing the item, signal the consumer thread.            
                _queueWaitHandle.Set();
            }
        }

        private void ConsumeItems(object state)
        {
            while (_isProcessing)
            {
                try
                {
                    var nextItem = default(T);
                    bool doesItemExist;
                    lock (_queueLocker)
                    {
                        int queueCount = _queue.Count;
                        doesItemExist = queueCount > 0;
                        if (doesItemExist)
                        {
                            nextItem = _queue.Dequeue();
                        }
                        if (queueCount > 0 && queueCount % 50 == 0)
                            _log4NetLogger.Warn(String.Format("Queue is/has been growing.  Queue size now:{0}",
                                                              queueCount));
                    }
                    if (doesItemExist)
                    {
                        _consumerAction(nextItem);
                    }
                    else
                    {
                        _queueWaitHandle.WaitOne();
                    }
                }
                catch (Exception ex)
                {

                    _log4NetLogger.Error(ex);
                }

            }
        }
    }

这是一个通用类，因此 T 是您要处理的对象类型。您还为其提供一个 Action，这是执行实际处理的方法。这应该允许您以干净的方式一次处理多个 PDF 文件。