Multithreading With C# Cookbook---Chapter3---线程池

概念

因为创建线程是昂贵的操作（线程管理/调度、上下文切换等操作相当耗时），如果为每个短暂的异步操作去创建新线程会明显增加开销。因此我们考虑只花费极少的时间来完成创建很多异步操作。叫做线程池（Thread pooling）。我们先分配一些资源到资源池，每次需要使用的时候从资源池获取，用完之后归还至线程池即可。

更多内容

基于.NET，通过System.Threading.ThreadPool类型使用线程池，受.NET通用语言运行时（Commom Language Runtime，CLR）管理。

TreadPool类型拥有一个QueueUserWorkItem静态方法，该方法接受一个委托，用户可自定义一个异步操作，当该方法被调用后，委托会进入内部队列。

这时，如果线程池没有任何线程，则创建一个新的工作者线程（worker thread），并将队列中的第一个委托放入该工作者线程。

如果线程池已有线程，且该线程处于闲置状态，则直接调用该线程；但如果线程池没有闲置的线程，则会创建更多的线程来执行操作。

但能创建的线程总数是有限制的。如果线程池线程数量已达上限，则新的操作会在队列中等待有工作者线程能够来执行它。

在停止向线程池中防止新操作时，线程池会删除已过期不用的线程，来释放系统资源。

注意：使用线程池线程执行的操作应该是执行运行时间短暂的。如果在线程池放入长时间运行的操作，将导致工作者线程一直忙碌，从而无法响应新操作，失去了线程池的意义。

使用线程池可以减少并行度耗费以及节省操作资源。使用较少线程但花费比每个操作都创建新线程更多的时候来执行异步操作，可以使用一定数量的工作者线程来批量处理所有操作。如果操作执行时间较短，则比较适合线程池；如果执行长时间运行的计算密集型操作则会降低性能。

线程池中的所有工作者线程都是后台线程。

本章所用using

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using static System.Threading.Thread;
using System.Threading;
using System.Diagnostics;
using System.ComponentModel;

在线程池中调用委托

Thread.CurrentThread.IsThreadPoolTread属性值来确认当前线程是否来自线程池。这里的code使用委托的BegainInvoke方法来运行委托，并提供回调函数，在异步操作完成之后调用；用IAsyncReasult接口对象来结束BegainInvoke结果，可以调用IAsyncReasult的IsCompleted属性轮询结果，也可以使用AsyncWaitHandle属性来等待操作完成。操作完成后通过委托调用EndInvoke方法来获取结果。（PS：其实直接调用EndInvoke，该方法会等待操作完成）

class Program
    {
        private delegate string RunOnThreadPool(out int threadId);
        static void Main(string[] args)
        {
            int threadId = 0;
            RunOnThreadPool poolDel = Test;
            var t = new Thread(() => Test(out threadId));
            t.Start();
            t.Join();
            Console.WriteLine("我是主线程:"+CurrentThread.ManagedThreadId);
            Console.WriteLine($"threadId:{threadId}");
            IAsyncResult ar = poolDel.BeginInvoke(out threadId, CallBack, "res a delegate async call");
            ar.AsyncWaitHandle.WaitOne();
            Console.WriteLine("我是主线程:" + CurrentThread.ManagedThreadId);
            string res = poolDel.EndInvoke(out threadId, ar);
            Console.WriteLine($" main threadPool worker threadId:{ threadId}");
            Console.WriteLine(res);

            Console.ReadLine();
        }

        private static void CallBack(IAsyncResult ar)
        {
            Console.WriteLine("callBack：starting……");
            Console.WriteLine($"callBack：State passed to a callBack:{ar.AsyncState}");
            Console.WriteLine($"callBack：CallBack is threadPool thread:{CurrentThread.IsThreadPoolThread}");
            Console.WriteLine($"callBack：CallBack threadPool worker threadId:{CurrentThread.ManagedThreadId}");
        }
        private static string Test(out int threadId)
        {
            Console.WriteLine("Starting Test……");
            Console.WriteLine($"is threadPool thread:{CurrentThread.IsThreadPoolThread}");
            Thread.Sleep(TimeSpan.FromSeconds(5));
            threadId = CurrentThread.ManagedThreadId;
            return $"threadPool worker threadId:{CurrentThread.ManagedThreadId}";
        }
    }

这里使用BegainOperationName/EndOperationName方法和.NET的IAsyncResult对象等方式被称为异步编程模型（Asynchronous Programming Model，APM），也称为异步方法，是.NET历史上第一个异步编程模式。在现代编程更推荐使用任务并行库（Task Parallel Library，TPL）来组织异步API。

闭包

使用Lambda表达式引用另一个C#对象的方式被称为闭包。闭包更灵活，允许向异步操作传递一个以上的对象而且这些对象具有静态类型。

向线程池中放入异步操作

代码有4个示例，主要看第4个示例，该示例采用闭包机制，允许传递一个以上对象。

class Program
    {

        static void Main(string[] args)
        {
            const int x = 1, y = 2;
            const string lambdaState = "lambda state 2";

            ThreadPool.QueueUserWorkItem(AsyncOperation);
            Thread.Sleep(TimeSpan.FromSeconds(3));

            ThreadPool.QueueUserWorkItem(AsyncOperation, "async state");
            Thread.Sleep(TimeSpan.FromSeconds(3));

            ThreadPool.QueueUserWorkItem(state =>
            {
                Console.WriteLine($"Operation state:{state}");
                Console.WriteLine($"Worker thread id:{CurrentThread.ManagedThreadId}");
                Thread.Sleep(TimeSpan.FromSeconds(3));
                //Console.WriteLine("---------------------");
            }, "lambda state 3");

            ThreadPool.QueueUserWorkItem(_ =>
            {
                Console.WriteLine($"Operation state:{x + y},{lambdaState}");
                Console.WriteLine($"Worker thread id:{CurrentThread.ManagedThreadId}");
                Thread.Sleep(TimeSpan.FromSeconds(3));
                //Console.WriteLine("---------------------");
            }, "lambda state 4");

            Console.ReadLine();
        }

        private static void AsyncOperation(object state)
        {
            Console.WriteLine($"Operation state:{state ?? "(null)"}");//相当于state==null?null:state
            Console.WriteLine($"Worker thread id:{CurrentThread.ManagedThreadId}");
            Thread.Sleep(TimeSpan.FromSeconds(2));
            //Console.WriteLine("---------------------");
        }
    }

线程池与并行度

测试面对大量异步操作的时候，使用线程池与创建大量单独线程的不同之处。

class Program
    {

        static void Main(string[] args)
        {
            int numsOfOperations = 500;
            Stopwatch sw = new Stopwatch();
            sw.Start();
            UseThreads(500);
            sw.Stop();
            Console.WriteLine(sw.ElapsedMilliseconds);

            sw.Reset();
            sw.Start();
            UseThreadPool(numsOfOperations);
            sw.Stop();
            Console.WriteLine(sw.ElapsedMilliseconds);

            Console.ReadLine();
        }

        static void UseThreads(int numsOfOperations)
        {
            Console.WriteLine("Use threads……");
            using (CountdownEvent _countDown = new CountdownEvent(numsOfOperations))
            {
                for (int i = 0; i < numsOfOperations; i++)
                {
                    Thread thread = new Thread(() =>
                    {
                        Console.WriteLine("Thread id:" + CurrentThread.ManagedThreadId);
                        Thread.Sleep(TimeSpan.FromSeconds(0.1));
                        _countDown.Signal();
                    });

                    thread.Start();
                }
                _countDown.Wait();
            }
        }

        static void UseThreadPool(int numsOfOperations)
        {
            Console.WriteLine("Use thread pool……");
            using (CountdownEvent _countDown = new CountdownEvent(numsOfOperations))
            {
                for (int i = 0; i < numsOfOperations; i++)
                {
                    ThreadPool.QueueUserWorkItem(_ =>
                   {
                       Console.WriteLine("Thread id:" + CurrentThread.ManagedThreadId);
                       Thread.Sleep(TimeSpan.FromSeconds(0.1));
                       _countDown.Signal();
                   });
                }
                _countDown.Wait();
            }
        }
    }

测试结果表明，面对一次性的500个操作，使用线程池，所开的线程以及占用的系统资源较少，但运行时间要比每个单独创建线程更长。简单说，线程池为操作系统节省了内存和线程数，但为此付出了更长的执行时间。

在线程池中取消异步操作

第一种通过轮询CancellationToken的IsCancellationRequested属性来确认是否取消；第二种通过CancellationToken.ThrowIfCancellationRequested来抛出OperationCanceledException异常，需要加try…catch来捕获；第三种通过CancellationToken.Register注册一个回调函数，当操作被取消时被调用，来修改是否取消的标记。

class Program
    {
        static void Main(string[] args)
        {
            using (CancellationTokenSource cts = new CancellationTokenSource())
            {
                CancellationToken ct = cts.Token;
                ThreadPool.QueueUserWorkItem(_ => AsyncOperation1(ct));
                Thread.Sleep(TimeSpan.FromSeconds(2));
                cts.Cancel();
            }

            using (CancellationTokenSource cts = new CancellationTokenSource())
            {
                CancellationToken ct = cts.Token;
                ThreadPool.QueueUserWorkItem(_ => AsyncOperation2(ct));
                Thread.Sleep(TimeSpan.FromSeconds(2));
                //cts.Cancel();
            }

            using (CancellationTokenSource cts = new CancellationTokenSource())
            {
                CancellationToken ct = cts.Token;
                ThreadPool.QueueUserWorkItem(_ => AsyncOperation3(ct));
                Thread.Sleep(TimeSpan.FromSeconds(2));
                cts.Cancel();
            }

            Console.ReadLine();
        }

        static void AsyncOperation1(CancellationToken ct)
        {
            Console.WriteLine("Starting first task……");
            for (int i = 0; i < 5; i++)
            {
                if (ct.IsCancellationRequested)
                {
                    Console.WriteLine("First task has been canceled");
                    return;
                }
                Sleep(TimeSpan.FromSeconds(1));
            }
            Console.WriteLine("First task has completed successfully");
        }

        static void AsyncOperation2(CancellationToken ct)
        {
            Console.WriteLine("Starting second task……");
            try
            {
                for (int i = 0; i < 5; i++)
                {
                    ct.ThrowIfCancellationRequested();
                    Sleep(TimeSpan.FromSeconds(1));
                }
                Console.WriteLine("Second task has completed successfully");
            }
            catch (OperationCanceledException)
            {

                Console.WriteLine("Second task has been canceled");
            }

        }

        static void AsyncOperation3(CancellationToken ct)
        {
            bool cancelFlag = false;
            ct.Register(() => cancelFlag = true);
            Console.WriteLine("Starting third task……");

            for (int i = 0; i < 5; i++)
            {
                if (cancelFlag)
                {
                    Console.WriteLine("Third task has been canceled");
                    return;
                }
                Sleep(TimeSpan.FromSeconds(1));
            }
            Console.WriteLine("Third task has completed successfully");
        }
    }

CancellationTokenSource和CancellationToken两个新类是在.NET4.0之后被引入，目前是实现异步操作的取消操作的事实标准。

在线程池中使用超时

 使用ThreadPool.RegisterWaitForSingleObject()方法，提供回调函数来判断是否超时或者正常完成。这里可以使用ManualRestEvent对象来发出任务完成信号，使用CancellationToken来发出任务取消信号。

class Program
    {
        static void Main(string[] args)
        {
            RunOperation(TimeSpan.FromSeconds(5));
            RunOperation(TimeSpan.FromSeconds(7));

            Console.ReadLine();
        }

        static void RunOperation(TimeSpan workerOperationTimeout)
        {
            using (ManualResetEvent mre = new ManualResetEvent(false))
            using (CancellationTokenSource cts = new CancellationTokenSource())
            {
                Console.WriteLine("Registering timeout operation……");
                var worker = ThreadPool.RegisterWaitForSingleObject(mre, (state, isTimeOut) => WorkerOperationWait(cts, isTimeOut), null, workerOperationTimeout, true);//注册一个等待委托，指定超时时间
                Console.WriteLine("Starting long task……");
                ThreadPool.QueueUserWorkItem(_ => WorkerOperation(cts.Token, mre));
                Sleep(workerOperationTimeout.Add(TimeSpan.FromSeconds(2)));
                worker.Unregister(mre);//注销RegisterWaitForSingleObject发出的等待操作
            }
        }
        static void WorkerOperation(CancellationToken ct, ManualResetEvent mre)
        {
            for (int i = 0; i < 6; i++)
            {
                if (ct.IsCancellationRequested)
                {
                    return;//isTimeOut=true
                }
                Sleep(TimeSpan.FromSeconds(1));
            }
            mre.Set();//isTimeOut=false
        }
        static void WorkerOperationWait(CancellationTokenSource cts, bool isTimeOut)
        {
            if (isTimeOut)
            {
                cts.Cancel();
                Console.WriteLine("Worker operation is timeout");
            }
            else
            {
                Console.WriteLine("Worker operation is completed success");
            }

        }
    }

使用计时器

指定第一次运行操作时间，以及运行该操作的周期；再通过Change方法改变第一次运行时间以及运行周期。

class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Please enter to stop the timer");
            DateTime start = DateTime.Now;
            _timer = new Timer(_ => TimerOperation(start), null, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2));
            ;

            try
            {
                Thread.Sleep(TimeSpan.FromSeconds(5));
                _timer.Change(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(4));
                Console.ReadLine();
            }
            catch (Exception)
            {

                throw;
            }
            finally
            {
                _timer.Dispose();
            }
        }

        static Timer _timer;
        static void TimerOperation(DateTime start)
        {
            TimeSpan elapsed = DateTime.Now - start;
            Console.WriteLine($"{elapsed.Seconds} seconds from {start}. Timer thread pool thread id:{CurrentThread.ManagedThreadId}");
        }
    }

 使用BackgroundWorker组件

使用事件，事件表示了一些通知的源或当通知到达时会有所响应的一系列订阅者。当有事件发生时，将调用相应的事件处理器。

class Program
    {
        static void Main(string[] args)
        {
            var bw = new BackgroundWorker();
            bw.WorkerReportsProgress = true;
            bw.WorkerSupportsCancellation = true;

            bw.DoWork += Worker_DoWork;
            bw.ProgressChanged += Worker_ProgressChanaged;
            bw.RunWorkerCompleted += Worker_Completed;

            bw.RunWorkerAsync();//调用DoWork

            Console.WriteLine("Press C to cancel work");
            do
            {
                if (Console.ReadKey(true).KeyChar == 'C')
                {
                    bw.CancelAsync();//bw.CancellationPending=true
                }
            } while (bw.IsBusy);
        }

        static void Worker_DoWork(object sender, DoWorkEventArgs e)
        {
            Console.WriteLine($"DoWork thread pool thread id:" + CurrentThread.ManagedThreadId);
            var bw = (BackgroundWorker)sender;
            for (int i = 1; i < 100; i++)
            {
                if (bw.CancellationPending)
                {
                    e.Cancel = true;
                    return;
                }
                if (i % 10 == 0)
                {
                    bw.ReportProgress(i);//调用ProgressChanged
                }
                Sleep(TimeSpan.FromSeconds(0.1));
            }
            e.Result = 42; //完成后调用RunWorkerCompleted
        }

        static void Worker_ProgressChanaged(object sender, ProgressChangedEventArgs e)
        {
            Console.WriteLine($"{e.ProgressPercentage}% completed. Progress thread pool thread id:{CurrentThread.ManagedThreadId}");
        }

        static void Worker_Completed(object sender, RunWorkerCompletedEventArgs e)
        {
            Console.WriteLine($"Completed thread pool thread id:{CurrentThread.ManagedThreadId}");
            if (e.Error != null)
            {
                Console.WriteLine($"Exception {e.Error.Message} has occured.");
            }
            else if (e.Cancelled)
            {
                Console.WriteLine($"Operation has been canceled.");
            }
            else
            {
                Console.WriteLine($"The answer is : {e.Result}");
            }
        }
    }

注：本文是在阅读《C#多线程编程实战》后所写，部分内容引用该书内容，这是一本不错的书，感谢！