3

我创建了一个用于优化 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为所有处理器使用该功能，即我想同时优化多个文件。换句话说，我想让所有处理器忙于优化。

请指导我怎么可能？

4

4 回答 4

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 文件。

于 2012-12-06T08:40:23.320 回答
0

检查此线程：每个核心的最佳线程数

如果您的线程方法csCommon.pdfFilesCompressAndMove非常消耗 CPU（我可以通过它的名称猜测），您应该为每个内核启动 1 个线程。您最好使用ThreadPool.QueueUserWorkItem，而不是手动创建线程，它将负责在内核之间生成线程。在您的情况下，据我了解，您有 8 个内核，因此您可以调用ThreadPool.QueueUserWorkItem(csCommon.pdfFilesCompressAndMove)8 次，并在您的一个线程完成时再次调用，使正在运行的线程总数等于 8。

于 2012-12-06T08:37:51.960 回答
0

我会使用ThreadPool，因为据我所知，它是由 .NET Framework 和操作系统以始终为目标系统创建最佳线程数的方式管理的。

于 2012-12-06T08:58:35.943 回答
0

我认为你最好的选择是从一些简单的事情开始，这样你就可以了解问题的性能特征。

List<string> items = GetListOfPdfFilesToProcess();
int numCores = 4;
int maxListChunkSize = (int)Math.Ceiling(items.Count / (double)numCores);
ManualResetEvent[] events = new ManualResetEvent[numCores];

for (int i = 0; i < numCores; i++)
{
    ThreadPool.QueueUserWorkItem(ProcessFiles, new object[]
    {
        items.Skip(i * maxListChunkSize).Take(maxListChunkSize).ToList(), events[i]
    });
}

WaitHandle.WaitAll(events);

....

private static void ProcessFiles(object state)
{
    object[] stateArray = (object[])state;
    List<string> filePaths = (List<string>)stateArray[0];
    ManualResetEvent completeEvent = (ManualResetEvent)stateArray[1];

    for (int i = 0; i < filePaths.Count; i++)
    {
        csCommon.pdfFilesCompressAndMove(your parameters);
    }

    completeEvent.Set();
}

这里的主要是将工作分成numCores块。这样，您应该能够充分利用所有 CPU 内核，同时保持一个非常简单的编程模型。

请记住，这不会进行任何错误处理 - 您需要处理它。csCommon.pdfFilesCompressAndMove如果无法处理文件，也可能需要考虑一下该怎么做。最简单的方法是记录错误并稍后检查，但如果您认为下次会成功，您可以尝试重新处理该文件。

您会注意到该state对象只是一个数组；如果您需要传递大量参数，ProcessFiles那么将这些参数包装到单个对象中并将其作为state.

编辑：

要从您的Tick活动中使用：

private void TimerTick(object sender, EventArgs e)
{
    //Disabling the timer will ensure the `TimerTick` method will not try to run
    //while we are processing the files. This covers the case where processing takes
    //longer than 2 minutes.
    timer.Enabled = false;

    //Run the first block of code in my answer.

    //Reenabling the timer will start the polling back up.
    timer.Enabled = true;
}

我还建议检查您必须处理的文件数量：如果没有，请重新启用计时器并返回。这将避免排队一堆实际上不做任何事情的操作。

于 2012-12-06T10:00:42.647 回答