如何在 .NET 3.5 中重用线程

Question

我有一个处理大块信息的子程序。为了利用整个 CPU，它将工作分成单独的线程。在所有线程都完成后，它就完成了。我读到创建和销毁线程会占用大量开销，因此我尝试使用线程池，但实际上运行速度比创建自己的线程慢。如何在程序运行时创建自己的线程然后继续重用它们？看到有人说做不到，但是线程池做到了，应该是可以的吧？

这是启动新线程/使用线程池的部分代码：

//initialization for threads
Thread[] AltThread = null;
if (NumThreads > 1)
    AltThread = new Thread[pub.NumThreads - 1];

do
{
    if (NumThreads > 1)
    {   //split the matrix up into NumThreads number of even-sized blocks and execute on separate threads
        int ThreadWidth = DataWidth / NumThreads;
        if (UseThreadPool) //use threadpool threads
        {
            for (int i = 0; i < NumThreads - 1; i++)
            {
                ThreadPool.QueueUserWorkItem(ComputePartialDataOnThread, 
                    new object[] { AltEngine[i], ThreadWidth * (i + 1), ThreadWidth * (i + 2) });
            }
            //get number of threads available after queue
            System.Threading.Thread.Sleep(0);
            int StartThreads, empty, EndThreads;
            ThreadPool.GetAvailableThreads(out StartThreads, out empty);
            ComputePartialData(ThisEngine, 0, ThreadWidth);

            //wait for all threads to finish
            do
            {
                ThreadPool.GetAvailableThreads(out EndThreads, out empty);
                System.Threading.Thread.Sleep(1);
            } while (StartThreads - EndThreads > 0);
        }
        else //create new threads each time (can we reuse these?)
        {
            for (int i = 0; i < NumThreads - 1; i++)
            {
                AltThread[i] = new Thread(ComputePartialDataOnThread);
                AltThread[i].Start(new object[] { AltEngine[i], ThreadWidth * (i + 1), ThreadWidth * (i + 2) });
            }
            ComputePartialData(ThisEngine, 0, ThreadWidth);

            //wait for all threads to finish
            foreach (Thread t in AltThread)
                t.Join(1000);
            foreach (Thread t in AltThread)
                if (t.IsAlive) t.Abort();
        }
    }
}

ComputePartialDataOnThread 只是将信息解包并调用 ComputePartialData。将要处理的数据在线程之间共享（它们不会尝试读取/写入相同的位置）。 AltEngine[] 是每个线程的独立计算引擎。

该操作使用线程池运行大约 10-20%。

Answer 1

这听起来像是一个相当普遍的需求，可以通过多线程生产者-消费者队列来解决。线程保持“活动”，并在将新工作添加到队列时发出工作信号。工作由委托（在您的情况下为 ComputePartialDataOnThread）表示，传递给委托的数据是排队的数据（在您的情况下为 ComputePartialDataOnThread 的参数）。有用的特性是管理工作线程和实际算法的实现是分开的。这是 p-c 队列：

public class SuperQueue<T> : IDisposable where T : class
{
    readonly object _locker = new object();
    readonly List<Thread> _workers;
    readonly Queue<T> _taskQueue = new Queue<T>();
    readonly Action<T> _dequeueAction;

    /// <summary>
    /// Initializes a new instance of the <see cref="SuperQueue{T}"/> class.
    /// </summary>
    /// <param name="workerCount">The worker count.</param>
    /// <param name="dequeueAction">The dequeue action.</param>
    public SuperQueue(int workerCount, Action<T> dequeueAction)
    {
        _dequeueAction = dequeueAction;
        _workers = new List<Thread>(workerCount);

        // Create and start a separate thread for each worker
        for (int i = 0; i < workerCount; i++)
        {
            Thread t = new Thread(Consume) { IsBackground = true, Name = string.Format("SuperQueue worker {0}",i )};
            _workers.Add(t);
            t.Start();
        }
    }

    /// <summary>
    /// Enqueues the task.
    /// </summary>
    /// <param name="task">The task.</param>
    public void EnqueueTask(T task)
    {
        lock (_locker)
        {
            _taskQueue.Enqueue(task);
            Monitor.PulseAll(_locker);
        }
    }

    /// <summary>
    /// Consumes this instance.
    /// </summary>
    void Consume()
    {
        while (true)
        {
            T item;
            lock (_locker)
            {
                while (_taskQueue.Count == 0) Monitor.Wait(_locker);
                item = _taskQueue.Dequeue();
            }
            if (item == null) return;

            // run actual method
            _dequeueAction(item);
        }
    }

    /// <summary>
    /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
    /// </summary>
    public void Dispose()
    {
        // Enqueue one null task per worker to make each exit.
        _workers.ForEach(thread => EnqueueTask(null));

        _workers.ForEach(thread => thread.Join());

    }
}

正如之前的海报所说，有许多内置结构（查看 TPL ），它们使用线程池，您可能希望在实现自己的队列之前查看它。

Answer 2

因此，通常的做法是让每个线程的入口点本质上执行类似的操作（这只是一种算法，不是 C# 代码，抱歉）：

1. 检查您是否有工作要做
2. 找到后就开始工作
3. 等待信号

另一方面，当您的线程有更多工作时，将其添加到工作队列中，然后您的线程实质上就会被重用。这与自己实现线程池的方式非常相似（如果您在运行时中，您可以做一些其他事情来帮助您，但这并不是什么大不了的事）。

Answer 3

这是一个讨论这个问题的线程：A custom thread-pool/queue class.