.NET Core多线程通关 Thread与Task

Posted dotNET跨平台

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了.NET Core多线程通关 Thread与Task相关的知识，希望对你有一定的参考价值。

【.NET Core】| 总结/Edison Zhou

大家好，我是Edison。

去年换工作时系统复习了一下.NET Core多线程相关专题，学习了一线码农老哥的《.NET 5多线程编程实战》课程，我将复习的知识进行了总结形成本专题。同时也特别推荐有兴趣的读者去学习一线码农老哥的《.NET 5多线程编程》课程。

本篇，我们来复习一下Thread与Task的相关知识点，预计阅读时间15~20分钟。

从时间和空间角度理解线程的开销

（1）多线程的优点

  • 提高响应能力

      • main thread：更新UI的东西

      • work thread：耗时的操作

  • 提高程序性能

      • 1个力工：1个月

      • 10个力工：3天~5天

（2）线程有哪些开销

  • 空间上的开销

    • 数据结构上的开销

      • C#：Thread

      • CLR：Thread（C++写的）

      • OS：Thread

    • 线程栈开销

      • 默认最大栈空间：1MB

      • 线程越多，栈空间越大

    • teb开销（thread enviornment block，线程环境块）

      • ThreadStatic

      • TLS

  • 时间上的开销

    • dllmain

      • 非托管dll 上面上游 dllmain

      • thread 在start时 会通知这些 非托管dll

      • thread 在exit时 也会通知这些非托管dll（资源清理）

    • switch context

      • Windows系统中大概30ms进行一次上下文切换，如果上下文切换非常频繁，会造成CPU暴高

      • 在上下文切换中涉及到CPU与thread的交互

        • 时间片到了，thread 暂停，涉及到数据保存（将高速缓存中的数据存到线程的本地存储中）

        • 时间片分配，thread 恢复，涉及到数据恢复（从线程的环境块中将当时的数据重新提取出来）

（3）总结

线程不是越多越好，线程有时间和空间上的开销，所以我们需要省着用。

线程的常用方法及生命周期管理

（1）Thread的基本操作

  • Start

    • 不带参数：new Thread(()=> xxxxxx ).Start();

    • 带参数：new Thread((obj)=> xxxxxx ).Start();

  • Join

    • 类似于Task.Wait()方法的作用

    • 不带超时参数：thread.Join();

    • 带超时参数：thread.Join(1000 * 5);

  • Sleep

    • 冻结当前线程指定时间：Thread.Sleep(1000 * 5);

  • IsBackground属性

    • 指明当前线程为 后台线程

    • 如果主线程退出，后台线程自动退出

    • 只有所有的前台线程都退出了，主线程才能退出

（2）对Thread的思考

现在实际开发中直接用thread的不多，因为它较为底层，很多程序员用不好。

  • 线程太多，造成上下文切换频繁（CPU暴高）

    • 比如创建了5000个thread，假设都在执行耗时任务，而运行主机只有6核12线程，必然会造成频繁的上下文切换

  • GC负担过大，徒增GC负担

    • 比如创建了5000个thread跑了任务后，虽然没有引用根了，但是GC还没有及时回收，因此这时它们就是dead thread，它们全都在托管堆上

（3）一些解决方案

  • ThreadPool：线程池

  • Task：基于ThreadPool的上层封装

线程池的使用及分析其设计思想

（1）为什么要使用线程池？

  • GC负担

  • 上下文切换

让thread得到更好的使用，提高利用率，减少不必要的创建和销毁。

（2）线程池的基本使用

  • 无参数

    • ThreadPool.QueueUserWorkItem(_ => ........... );

  • object参数

    • ThreadPool.QueueUserWorkItem(obj => p = obj as Person; ........... , new Person() Name = "Edison" );

    • 由于是object类型，涉及到多余的类型转换

  • 泛型参数

    • ThreadPool.QueueUserWorkItem

      (p => ........... , new Person() Name = "Edison" , true);

    • 第三个参数 bool preferLocal，一般建议传true，代表优先使用线程本地队列（Local Queue） 而不是 全局队列（Work Queue），降低锁竞争。

  • 其他方法

    • GetMinThreads, GetMaxThreads

    • ThreadCount、CompletedWorkItemCount

（3）ThreadPool的设计

  • WinDbg视角下的ThreadPool

  • ThreadPool的设计图如下：

在老版本的.NET Framework时代，只有一个全局队列，存在大量的锁竞争。

.NET Core中加入了本地队列，加入了本地队列，降低了锁竞争，并提高了线程的利用率。

具体实现思路是：

（1）每个线程优先从本地队列中取任务干活；

（2）如果本地队列中没有任务了，就从全局队列中取任务干活；

（3）当全局任务队列里面的任务没有的时候，CLR将会把其他有任务的线程中的未处理任务（比如上图中的WorkItem3），分配给这些空闲的线程（比如上图中的Thread3）去执行。这个机制也被称之为 偷窃机制。

这样做的其目的是每个线程都有事干，即提高线程池中的线程利用率。

Task及如何运用其编排能力

（1）Task的设计思想

为什么会出现Task：

  • 获取Thread的返回值比较麻烦

  • 多个Thread的串行实现比较麻烦

  • Thread的父子关系实现比较麻烦（比如：所有的子Thread执行完后，才能结束父Thread）

本质问题：如何高效地对Thread进行编排？

本质理解：Task就是一个Thread的编排工具，它解决了任务之间如何串行、如何并行、如何嵌套、如何父子等关系的处理，让程序员可以重点关注任务，而不是Thread。

（2）Task的基本使用

方式一：new Task，不推荐使用

// 无参数
var task = new Task(()=>

    Console.WriteLine($"Current ThreadId=Environment.CurrentManagedThreadId");
);
task.Start();
// 有参
var task = new Task((obj)=>

    Console.WriteLine($"Current ThreadId=Environment.CurrentManagedThreadId, Current Content=obj");
, "Hello World");
task.Start()

方式二：Task.Factory.StartNew

// 无参数
var task = Task.Factory.StartNew(()=>

    Console.WriteLine($"Current ThreadId=Environment.CurrentManagedThreadId");
);
// 有参
var task = Task.Factory.StartNew((obj)=>

    Console.WriteLine($"Current ThreadId=Environment.CurrentManagedThreadId, Current Content=obj");
, "Hello World");

方式三：Task.Run

// 无参数
var task = Task.Run(()=>

    Console.WriteLine($"Current ThreadId=Environment.CurrentManagedThreadId");
);
// 有参
var task = Task.Run((obj)=>

    Console.WriteLine($"Current ThreadId=Environment.CurrentManagedThreadId, Current Content=obj");
, "Hello World");

Task串行、父子、并行等玩法

（1）串行玩法

var task1 = Task.Factory.StartNew(()=>

    new Sheet1().WriteSheet();
).ContinueWith(t =>

    new Sheet2().WriteSheet();
).ContinueWith(t =>

    new Sheet0().WriteSheet();
);
task1.Wait();

（2）并行+串行玩法

var sheets = new List<Sheet>  new Sheet1(), new Sheet2() ;
var tasks = new Task[2];
for(int i=0; i<sheets.Count; i++)

    tasks[i] = Task.Factory.StartNew((index)=>
    
        sheets[(int)index].WriteSheet();
    , i);

Task.WhenAll(tasks).ContinueWith(t=>

    new Sheet[0].WriteSheet();
).Wait();

（3）父子关系玩法

如果父Task中的任意一个子Task未完成，都不能继续。注意点：参数TaskCreationOptions.AttachedToParent

var sheets = new List<Sheet>  new Sheet1(), new Sheet2() ;


//父task
var parent_task = Task.Factory.StartNew(() =>

    //1. 子task1
    var child_1_task = Task.Factory.StartNew(() =>
    
        new Sheet1().WriteSheet();
    , TaskCreationOptions.AttachedToParent);


    //2. 子task2
    var child_2_task = Task.Factory.StartNew(() =>
    
        new Sheet2().WriteSheet();
    , TaskCreationOptions.AttachedToParent);
);


var continueTask= parent_task.ContinueWith(t =>

    new Sheet0().WriteSheet();
);




Task.WhenAll(continueTask);

最后等待可以有几种写法：

continueTask.Wait();
Task.WaitAll(continueTask);
Task.WaitAny(continueTask);

以上三种会阻塞主线程。

Task.WhenAll(continueTask);

上面这种方式不会阻塞主线程。

解析：WaitAll/WaitAny方法阻塞了当前线程直到全完。WhenAll方法会开启个新监控线程去判读括号里的所有线程执行情况并立即返回，等都完成了就退出监控线程并返回监控数据。

任务取消CTS机制的使用

CTS = CancellationTokenSource，它主要是帮助开发者实现优雅退出（Graceful Exit）。

（1）没有CTS之前如何处理的

一是Thread.Abort()

二是增加临时变量如isStop来判断（hard cod）

（2）理解框架中的CTS使用

namespace EDT.MultiThread.Demo

    class Program
    
        static void Main(string[] args)
        
            CTSDemo();
        


        static void CTSDemo()
        
            var source = new CancellationTokenSource();


            var task = Task.Factory.StartNew(() =>
            
                for (int i = 0; i < 5; i++)
                
                    Console.WriteLine($"当前线程:Environment.CurrentManagedThreadId, DateTime.Now 执行时间需要5s");


                    Thread.Sleep(1000);
                
            ).ContinueWith(t =>
            
                Console.WriteLine($"当前线程:Environment.CurrentManagedThreadId, 我是延续任务！");
            , source.Token);


            Thread.Sleep(3000);
            source.Cancel();


            Console.WriteLine("主线程要取消你啦。。");
            Console.ReadLine();
        


        /// <summary>
        /// 业务方法
        /// </summary>
        /// <param name="token"></param>
        static void Run(CancellationToken token)
        
            while (!token.IsCancellationRequested)
            
                Thread.Sleep(1000);
                Console.WriteLine("1. 正在处理 redis 业务");


                Thread.Sleep(1000);
                Console.WriteLine("2. 正在处理 mongodb 业务");


                Thread.Sleep(1000);
                Console.WriteLine("3. 正在处理 sqlserver 业务");


                Thread.Sleep(1000);
                Console.WriteLine("4. 正在处理 mysql 业务");

（3）其他功能

  • 延迟取消 CancelAfter

      • source.CancelAfter(1000 * 5);

  • 注册取消通知 Register

      • source.Token.Register(() => .......);

任务调度机制及其自定义

（1）TaskScheduler是什么

TaskScheduler决定了将Task调度到什么地方去执行，即TaskScheduler决定了Task如何被调度。

（2）BCL中现存的TaskScheduler

  • ThreadPoolTaskScheduler

      • 如果不特别指定，默认就是 ThreadPoolTaskScheduler

      • 内部有两种处理逻辑，一种是针对LongRunning需求的Task，会单独走后台Thread路径；另一种是非LongRunning需求的Task，直接走ThreadPool线程池路径。

      • Why？针对LongRunning的Task，如果长时间运行占用着ThreadPool的线程，这时候ThreadPool为了保证线程充足，会再次开辟一些Thread，如果耗时任务此时释放了，会导致ThreadPool线程过多，上下文切换频繁，所以这种情况下让Task在Thread中执行还是非常不错的选择

  • SynchronizationContextTaskScheduler

      • 适用于GUI程序：耗时操作一般不会放到UI线程处理，而是放到工作线程去处理，处理完之后通过发送消息到Queue，GUI程序就可以从Queue中取出来消费，更新UI内容。

      • How？在Task.Factory.StartNew方法中传参：TaskScheduler.FromCurrentSynchronizationContext()

（3）自己实现一个TaskScheduler

自己实现一个单个Thread处理所有Task的TaskScheduler：

namespace EDT.MultiThread.Demo

    public class CustomTaskScheduler : TaskScheduler
    
        Thread thread = null;


        BlockingCollection<Task> collection = new BlockingCollection<Task>();


        public CustomTaskScheduler()
        
            thread = new Thread(() =>
            
                foreach (var task in collection.GetConsumingEnumerable())
                
                    TryExecuteTask(task);
                
            );


            thread.Start();
        


        protected override IEnumerable<Task> GetScheduledTasks()
        
            return collection.ToArray();
        


        protected override void QueueTask(Task task)
        
            collection.Add(task);
        


        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
        
            throw new NotImplementedException();

调用端示例代码：

var scheduler = new CustomTaskScheduler();
for (int i = 0; i < 100; i++)

    var task = Task.Factory.StartNew(() =>
    
        Console.WriteLine($"当前线程:Environment.CurrentManagedThreadId");
    , CancellationToken.None, TaskCreationOptions.None, scheduler);



Console.ReadLine();

小结

本篇，我们复习了Thread与Task的基础知识。

下一篇，我们复习面试常考的重点-异步（async/await）相关知识。

参考资料

一线码农，腾讯课堂《.NET 5多线程编程实战》

不明作者，《Task调度与await》

年终总结：Edison的2022年终总结

数字化转型：我在传统企业做数字化转型

C#刷题：C#刷剑指Offer算法题系列文章目录

.NET面试：.NET开发面试知识体系

.NET大会：2020年中国.NET开发者大会PDF资料

以上是关于.NET Core多线程通关 Thread与Task的主要内容，如果未能解决你的问题，请参考以下文章

.NET Core多线程通关 常见性能问题

.NET Core多线程通关 锁机制

.NET Core多线程通关 异步 - 续

.NET Core 多线程的用法,以及用例

day10-02_多线程之进程与线程的pid

JAVA面试通关要点－基础篇－线程