NET线程池

C#并行编程（2）：.NET线程池

 

线程 Thread

在总结线程池之前，先来看一下.NET线程。

.NET线程与操作系统(Windows)线程有什么区别？

.NET利用Windows的线程处理功能。在C#程序编写中，我们首先会新建一个线程对象System.Threading.Thread，并为其指定一个回调方法；当我们调用线程对象的Start方法启动线程时，会创建一个操作系统线程来执行回调方法。.NET中的线程实际上等价于Windows系统线程，都是CPU调度和分配的对象。

前台线程和后台线程

.NET把线程分为前台线程和后台线程，两者几乎相同，唯一的区别是，前台线程会阻止进程的正常退出，后台线程则不会。下面用一个例子描述前、后台线程的区别：

class Program
{
    static void Main(string[] args)
    {
        ThreadDemo threadDemo = new ThreadDemo();//间隔500ms持续打印1-20至控制台
        //threadDemo.RunForegroundThread(); //用前台线程打印
        threadDemo.RunBackgroundThread(); //用台线程打印

        {//把主线程挂起5s，然后终止主线程
            Thread.Sleep(5000);
            Thread.CurrentThread.Abort();
        }//如果用前台线程工作，打印到20才会终止；以后台线程，则不会打印到20

        Console.ReadKey();
    }
}

public class ThreadDemo
{
    private readonly Thread _foregroundThread;
    private readonly Thread _backgroundThread;

    public ThreadDemo()
    {
        this._foregroundThread = new Thread(WriteNumberWorker) { Name = "ForegroundThread"};//默认是前台线程
        this._backgroundThread = new Thread(WriteNumberWorker) { Name = "BackgroundThread", IsBackground = true };//后台线程
    }

    /// <summary>
    /// 模拟线程工作
    /// </summary>
    private static void WriteNumberWorker()
    {
        for (int i = 0; i < 20; i++)
        {
            Console.WriteLine($"{DateTime.Now}=> {Thread.CurrentThread.Name} writes {i + 1}.");//打印从1到20到控制台
            Thread.Sleep(500);//挂起500毫秒，方便测试
        }
    }

    /// <summary>
    /// 运行前台线程
    /// </summary>
    public void RunForegroundThread()
    {
        this._foregroundThread?.Start();
    }

    /// <summary>
    /// 运行后台线程
    /// </summary>
    public void RunBackgroundThread()
    {
        this._backgroundThread?.Start();
    }
}

线程池 ThreadPool

线程的创建和销毁要耗费很多时间，而且过多的线程不仅会浪费内存空间，还会导致线程上下文切换频繁，影响程序性能。为改善这些问题，.NET运行时（CLR）会为每个进程开辟一个全局唯一的线程池来管理其线程。

线程池内部维护一个操作请求队列，程序执行异步操作时，添加目标操作到线程池的请求队列；线程池代码提取记录项并派发给线程池中的一个线程；如果线程池中没有可用线程，就创建一个新线程，创建的新线程不会随任务的完成而销毁，这样就可以避免线程的频繁创建和销毁。如果线程池中大量线程长时间无所事事，空闲线程会进行自我终结以释放资源。

线程池通过保持进程中线程的少量和高效来优化程序的性能。

C#中线程池是一个静态类，维护两种线程，工作线程和异步IO线程，这些线程都是后台线程。线程池不会影响进程的正常退出。

线程池的使用

线程池提供两个静态方法SetMaxThreads和SetMinThreads让我们设置线程池的最大线程数和最小线程数。最大线程数指的是，该线程池能够创建的最大线程数，当线程数达到设定值且忙碌，异步任务将进入请求队列，直到有线程空闲才会执行；最小线程数指的是，线程池优先尝试以设置数量的线程处理请求，当请求数达到一定量（未做深入研究）时，才会创建新的线程。

下面的例子展示了线程池的特性及常见使用方式。

class Program
{
    static void Main(string[] args)
    {
        //RunThreadPoolDemo();
        RunCancellableWork();

        Console.ReadKey();
    }

    static void RunThreadPoolDemo()
    {
        ThreadPoolDemo.ThreadPoolDemo.ShowThreadPoolInfo();
        ThreadPool.SetMaxThreads(100, 100); // 默认(1023,1000)(8核心CPU)
        ThreadPool.SetMinThreads(8, 8); // 默认是CPU核心数
        ThreadPoolDemo.ThreadPoolDemo.ShowThreadPoolInfo();
        ThreadPoolDemo.ThreadPoolDemo.MakeThreadPoolDoSomeWork(100);//计算限制任务
        ThreadPoolDemo.ThreadPoolDemo.MakeThreadPoolDoSomeIOWork();//IO限制任务
    }

    static void RunCancellableWork()
    {
        Console.WriteLine($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] started a work");
        Console.WriteLine("Press 'Esc' to cancel the work.");
        Console.WriteLine();
        ThreadPoolDemo.ThreadPoolDemo.DoSomeWorkWithCancellation();
        if (Console.ReadKey(true).Key == ConsoleKey.Escape)
        {// 发送取消通知
            ThreadPoolDemo.ThreadPoolDemo.CTSource.Cancel();
        }
    }
}

public class ThreadPoolDemo
{
    /// <summary>
    /// 显示线程池信息
    /// </summary>
    public static void ShowThreadPoolInfo()
    {
        int workThreads, completionPortThreads;

        //当前线程池可用的工作线程数量和异步IO线程数量
        ThreadPool.GetAvailableThreads(out workThreads, out completionPortThreads);
        Console.WriteLine($"GetAvailableThreads => workThreads:{workThreads};completionPortThreads:{completionPortThreads}");
        //线程池最大可用的工作线程数量和异步IO线程数量
        ThreadPool.GetMaxThreads(out workThreads, out completionPortThreads);
        Console.WriteLine($"GetMaxThreads => workThreads:{workThreads};completionPortThreads:{completionPortThreads}");
        //出现新的请求，判断是否需要创建新线程的依据
        ThreadPool.GetMinThreads(out workThreads, out completionPortThreads);
        Console.WriteLine($"GetMinThreads => workThreads:{workThreads};completionPortThreads:{completionPortThreads}");
        Console.WriteLine();
    }

    /// <summary>
    /// 让线程池做些事情
    /// </summary>
    public static void MakeThreadPoolDoSomeWork(int workCount = 10)
    {
        for (int i = 0; i < workCount; i++)
        {
            int index = i;
            // 将方法排队进入线程池工作项队列，当线程池有空闲线程时执行方法
            ThreadPool.QueueUserWorkItem(s =>
            {
                Thread.Sleep(100); //模拟工作时长
                Debug.Print($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] is running. [{index}]");
                ShowAvailableThreads("WorkerThread");
            });
        }
    }

    /// <summary>
    /// 让线程池做些IO工作
    /// </summary>
    public static void MakeThreadPoolDoSomeIOWork()
    {
        //随便找一些可以访问的网址
        IList<string> uriList = new List<string>()
        {
            "http://news.baidu.com/",
            "https://www.hao123.com/",
            "https://map.baidu.com/",
            "https://tieba.baidu.com/",
            "https://wenku.baidu.com/",
            "http://fanyi-pro.baidu.com",
            "http://bit.baidu.com/",
            "http://xueshu.baidu.com/",
            "http://www.cnki.net/",
            "http://www.wanfangdata.com.cn",
        };

        foreach (string uri in uriList)
        {
            WebRequest request = WebRequest.Create(uri);
            request.BeginGetResponse(ac =>
            {// 异步请求网址，将会利用线程池中异步IO线程
                try
                {
                    WebResponse response = request.EndGetResponse(ac);
                    ShowAvailableThreads("IOThread");
                    Debug.Print($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] is running. [{response.ContentLength}]");
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
            }, request);
        }
    }

    /// <summary>
    /// 打印线程池当前可用线程数
    /// </summary>
    private static void ShowAvailableThreads(string sourceTag = null)
    {
        int workThreads, completionPortThreads;
        ThreadPool.GetAvailableThreads(out workThreads, out completionPortThreads);
        Console.WriteLine($"{sourceTag} GetAvailableThreads => workThreads:{workThreads};completionPortThreads:{completionPortThreads}");
        Console.WriteLine();
    }

    /// <summary>
    /// 取消通知者
    /// </summary>
    public static CancellationTokenSource CTSource { get; set; } = new CancellationTokenSource();

    /// <summary>
    /// 执行可取消的任务
    /// </summary>
    public static void DoSomeWorkWithCancellation()
    {
        ThreadPool.QueueUserWorkItem(t =>
        {
            Console.WriteLine($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] begun running. [0 - 9999]");

            for (int i = 0; i < 10000; i++)
            {
                if (CTSource.Token.IsCancellationRequested)
                {
                    Console.WriteLine($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] recived the cancel token. [{i}]");
                    break;
                }
                Thread.Sleep(100);// 模拟工作时长
            }

            Console.WriteLine($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] was cancelled.");
        });
    }
}

线程池的调度

前面提到，线程池内部维护者一个工作项队列，这个队列指的是线程池全局队列。实际上，除了全局队列，线程池会给每个工作者线程维护一个本地队列。

当我们调用ThreadPool.QueueUserWorkItem方法时，工作项会被放入全局队列；使用定时器Timer的时候，也会将工作项放入全局队列；但是，当我们使用任务Task的时候，假如使用默认的任务调度器，任务会被调度到工作者线程的本地队列中。

工作者线程优先执行本地队列中最新进入的任务，如果本地队列中已经没有任务，线程会尝试从其他工作者线程任务队列的队尾取任务执行，这里需要进行同步。如果所有工作者线程的本地队列都没有任务可以执行，工作者线程才会从全局队列取最新的工作项来执行。所有任务执行完毕后，线程睡眠，睡眠一定时间后，线程醒来并销毁自己以释放资源。

线程池处理异步IO的内部原理

上面的例子中，从网站获取信息需要用到线程池的异步IO线程，线程池内部利用IOCP（IO完成端口）与硬件设备建立连接。异步IO实现过程如下：

1. 托管的IO请求线程调用Win32本地代码ReadFile方法
2. ReadFile方法分配IO请求包IRP并发送至Windows内核
3. Windows内核把收到的IRP放入对应