.NET Core多线程 (1) Thread与Task

Edison Zhou

发布于 2023-08-09 16:29:43

3070

发布于 2023-08-09 16:29:43

文章被收录于专栏：EdisonTalkEdisonTalk

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

（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();