等待线程池中所有任务完成

Question

我需要有关此代码的帮助，因此如果有人愿意提供帮助，我将不胜感激。

我正在开发一个简单的线程池，它传递一个数字数组，线程并行计算每个数字的阶乘。我的代码有什么问题，例如，如果我有两个线程并且有六个任务怎么办？两个线程接受一个任务，完成它，然后程序就结束了。我想要的是让他们接受任务，去做，如果还有更多任务，开始新的任务并做，只有当所有任务完成时程序才结束。

这是我的代码：

MyThreadPool.cs：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;
using System.Diagnostics;

namespace ThreadPool
{
    public class MyThreadPool : IDisposable
    {
        private bool _disposed = false;

        private object _lock = new object();

        private Queue<MyTask> tasks = new();

        public MyThreadPool(int numThreads = 0)
        {
            for (int i = 0; i < numThreads; i++)
            {
                new Thread(() =>
                {
                    while (_disposed == false)
                    {

                        MyTask? task;
                        lock (_lock)
                        {
                            while (!tasks.TryDequeue(out task) && _disposed == false)
                            {
                                Monitor.Wait(_lock);
                            }
                        }
                        if (task != null)
                        {
                            Console.WriteLine(@$"Thread {Thread.CurrentThread
                                .ManagedThreadId} calculated factorial for {task
                                .getNum()}. Result: {task.CalculateFactorial(task
                                .getNum())}");
                            lock (_lock)
                            {
                                if (tasks.Count > 0)
                                {
                                    Monitor.Pulse(_lock);
                                }
                            }
                        }
                    }
                }).Start();
            }
        }

        public MyTask Enqueue(int number)
        {
            MyTask task = new(number);
            lock (_lock)
            {
                tasks.Enqueue(task);
                Monitor.Pulse(_lock);
            }
            return task;
        }

        public void Dispose()
        {
            _disposed = true;
            lock (_lock)
            {
                Monitor.PulseAll(_lock);
            }
        }
    }
}

我的任务.cs：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

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 CalculateFactorial(int number)
        {
            int result = 1;
            for (int i = number; i > 0; i--)
            { 
                result = result * i;
            }
            return result;
        }
    }
}

程序.cs：

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

threadPool.Dispose();

Answer 1

方法有很多，比如添加计数。

private int _incompleteTaskCount = 0;

public MyTask Enqueue(int number)
{
    ....
    lock (_lock)
    {
        tasks.Enqueue(task);
        _incompleteTaskCount++;
        Monitor.Pulse(_lock);
    }
    ....
}

public MyThreadPool(int numThreads = 0)
{
    ....
    lock (_lock)
    {
        _incompleteTaskCount--;
        // if (tasks.Count > 0)
        // {
            Monitor.Pulse(_lock);
        // }
    }
    ....
}

然后你就可以等到

_incompleteTaskCount

变成0。

public void WaitForComplete()
{
    lock (_lock)
    {
        while (_incompleteTaskCount > 0)
        {
            Monitor.Wait(_lock);
        }
    }
}