c# - 如何在 .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

这听起来像是一个相当普遍的需求，可以通过多线程生产者-消费者队列来解决。线程保持“事件”状态，并在新工作添加到队列时发出工作信号。工作由委托(delegate)(在您的情况下是 ComputePartialDataOnThread)表示，传递给委托(delegate)的数据是排队的数据(在您的情况下是 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)，它们使用线程池，在实现您自己的队列之前您可能需要了解一下。