自定义ThreadPool线程唤醒问题求助：线程无法持续执行任务

问题描述 #

我编写了一个简易的自定义ThreadPool，功能是接收数字数组，让线程池中的线程为每个数字计算阶乘。当前遇到的问题是：当创建包含2个线程的线程池时，仅前2个任务能被执行，后续任务无法触发线程唤醒继续执行。不清楚如何实现线程的重新唤醒，以及让线程在有任务时持续执行的逻辑。

原代码 #

MyThreadPool类 #

using System;
using System.Collections.Generic;
using System.Threading;

namespace ThreadPool
{
    public class MyThreadPool
    {
        private bool _disposed = false;
        private object _lock = new object();
        private Queue<MyTask> tasks = new();
        private List<Thread> threads = new();
        public MyThreadPool(int numThreads = 0)
        {
            for (int i = 0; i < numThreads; i++)
            {
                new Thread(() =>
                {
                    threads.Add(Thread.CurrentThread);
                    if (_disposed == false)
                    {
                        MyTask task = null;
                        lock (_lock)
                        {
                            if (tasks.Count > 0)
                            {
                                task = tasks.Dequeue();
                            }
                            else
                                _disposed = true;

                            if (task != null)
                            {
                                Console.WriteLine($"Thread {Thread.CurrentThread.ManagedThreadId} calculated factorial for {task.getNum()}. Result: {task.CalculateFactoriel(task.getNum())}");
                                Thread.Sleep(1000);
                            }
                            else
                            {
                                Thread.Sleep(1000);
                            }
                        }
                    }
                }).Start();
            }
        }
        public void Enqueue(int number)
        {
            tasks.Enqueue(new MyTask(number));
        }
        public void Dispose()
        {
            if (_disposed == true)
            {
                foreach (Thread thread in threads)
                {
                    thread.Join();
                }
            }
        }
    }
}

MyTask类 #

using System;

namespace ThreadPool
{
    public class MyTask
    {
        private int num;
        public void setNum(int num)
        {
            this.num = num;
        }
        public int getNum()
        {
            return this.num;
        }
        public MyTask(int number)
        {
            this.num = number;
        }
        public int CalculateFactoriel(int number)
        {
            int result = 1;
            for (int i = number; i > 0; i--)
            { 
                result = result * i;
            }
            return result;
        }
    }
}

Program.cs #

int[] array = { 1, 2, 3, 4, 5, 6 };
int numThread = 2;
int x = 0;
MyThreadPool threadPool = new(numThread);
for (int i = 0; i < array.Length; i++)
{
    threadPool.Enqueue(array[i]);
}
while (x != array.Length) ;

threadPool.Dispose();

问题分析 #

1. 线程仅执行一次任务：工作线程的逻辑只在启动时尝试获取一次任务，没有循环监听任务队列，执行完一个任务后直接退出
2. 无任务唤醒机制：Enqueue方法仅将任务加入队列，没有通知等待的线程有新任务可用
3. _disposed标记错误触发：当任务队列为空时直接设置_disposed = true，导致线程直接终止，无法处理后续入队的任务
4. Dispose方法逻辑错误：仅当_disposed为true时才等待线程结束，但实际应该先标记关闭线程池，再等待线程完成剩余任务
5. Program.cs无限循环：变量x始终为0，导致程序永远无法进入Dispose逻辑

修复方案 #

修复后的MyThreadPool类 #

using System;
using System.Collections.Generic;
using System.Threading;

namespace ThreadPool
{
    public class MyThreadPool : IDisposable
    {
        private bool _isShuttingDown = false;
        private readonly object _lock = new object();
        private readonly Queue<MyTask> _tasks = new();
        private readonly List<Thread> _workers = new();
        private int _completedTasks = 0;
        private readonly int _totalTasks;

        // 新增构造参数接收总任务数，用于外部判断完成状态
        public MyThreadPool(int numThreads, int totalTasks)
        {
            _totalTasks = totalTasks;
            for (int i = 0; i < numThreads; i++)
            {
                var worker = new Thread(WorkerLoop);
                _workers.Add(worker);
                worker.Start();
            }
        }

        private void WorkerLoop()
        {
            while (true)
            {
                MyTask task = null;
                lock (_lock)
                {
                    // 等待直到有任务，或者线程池关闭且任务为空
                    while (_tasks.Count == 0 && !(_isShuttingDown && _tasks.Count == 0))
                    {
                        Monitor.Wait(_lock);
                    }

                    // 如果关闭且任务为空，退出循环
                    if (_isShuttingDown && _tasks.Count == 0)
                        break;

                    // 获取任务
                    task = _tasks.Dequeue();
                }

                // 执行任务（放在锁外，避免长时间占用锁）
                if (task != null)
                {
                    var result = task.CalculateFactoriel(task.getNum());
                    Console.WriteLine($"Thread {Thread.CurrentThread.ManagedThreadId} calculated factorial for {task.getNum()}. Result: {result}");
                    Thread.Sleep(1000);

                    // 标记任务完成
                    Interlocked.Increment(ref _completedTasks);
                }
            }
        }

        public void Enqueue(int number)
        {
            lock (_lock)
            {
                _tasks.Enqueue(new MyTask(number));
                // 唤醒一个等待的线程
                Monitor.Pulse(_lock);
            }
        }

        public bool AreAllTasksCompleted()
        {
            return _completedTasks >= _totalTasks;
        }

        public void Dispose()
        {
            lock (_lock)
            {
                _isShuttingDown = true;
                // 唤醒所有等待的线程，让它们退出循环
                Monitor.PulseAll(_lock);
            }

            // 等待所有工作线程结束
            foreach (var worker in _workers)
            {
                worker.Join();
            }
        }
    }
}

修复后的Program.cs #

int[] array = { 1, 2, 3, 4, 5, 6 };
int numThreads = 2;
MyThreadPool threadPool = new(numThreads, array.Length);

foreach (int num in array)
{
    threadPool.Enqueue(num);
}

// 等待所有任务完成
while (!threadPool.AreAllTasksCompleted())
{
    Thread.Sleep(100);
}

threadPool.Dispose();
Console.WriteLine("All tasks completed.");

修复要点说明 #

1. 循环监听任务队列：工作线程通过WorkerLoop方法持续循环，直到线程池关闭且任务为空
2. 线程等待与唤醒：使用Monitor.Wait让线程在任务队列为空时进入等待状态，Enqueue时用Monitor.Pulse唤醒等待的线程；关闭线程池时用Monitor.PulseAll唤醒所有线程
3. 修正关闭逻辑：用_isShuttingDown标记线程池是否关闭，只有调用Dispose时才设置该标记，线程会在完成剩余任务后退出
4. 任务完成跟踪：通过_completedTasks和AreAllTasksCompleted方法让外部可以判断所有任务是否执行完毕，避免无限循环
5. 锁范围优化：任务执行逻辑放在锁外部，只在操作任务队列时持有锁，提升并发效率

内容的提问来源于stack exchange，提问作者SP222