Task对于.NET的重要性毋庸置疑。通过最近的一些面试经历，发现很多人对与Task及其调度机制，以及线程和线程池之间的关系并没有清晰的认识。本文采用最简单的方式模拟了Task的实现，旨在说明Task是什么？它是如何被调度执行的？
Task对于.NET的重要性毋庸置疑。通过最近的一些面试经历，发现很多人对与Task及其调度机制，以及线程和线程池之间的关系并没有清晰的认识。本文采用最简单的方式模拟了Task的实现，旨在说明Task是什么？它是如何被调度执行的？源代码从这里下载。
一、Task（Job）一、Task（Job）
二、TaskScheduler（JobScheduler）
三、基于线程池的调度
四、使用指定线程进行调度
五、异步等待
六、await关键字的运用
七、状态机
Task代表一项具有某种状态的操作，我们使用如下这个Job类型来模拟Task。Job封装的操作体现为一个Action委托，状态则通过JobStatus枚举来表示（对应TaskStatus枚举）。简单起见，我们仅仅定义了四种状态（创建、调度、执行和完成）。Invoke方法负责执行封装的Action委托，并对状态进行相应设置。
二、TaskScheduler（JobScheduler）public class Job { private readonly Action _work; public Job(Action work)=> _work = work; public JobStatus Status { get; internal set; } internal protected virtual void Invoke() { Status = JobStatus.Running; _work(); Status = JobStatus.Completed;
} } public enum JobStatus { Created, Scheduled, Running, Completed }
Task承载的操作通过调度得以执行，具体的调度策略取决于调度器的选择。Task调度器通过TaskScheduler表示，我们利用如下这个JobScheduler类型对它进行模拟。如下面的代码片段所示，我们只为抽象类JobScheduler定义了唯一的QueueJob方法来调度作为参数的Job对象。静态Current属性表示当前默认实现的调度器。
public abstract class JobScheduler { public abstract void QueueJob(Job job); public static JobScheduler Current { get; set; } = new ThreadPoolJobScheduler (); }
对于开发者来说，执行Task就是将它提交给调度器，这一操作体现在我们为Job类型定义的静态Start方法中。该方法通过参数指定具体的调度器，如果没有显式指定，默认采用JobScheduler的Current静态属性设置的默认调度器。为了方便后面的演示，我们还定义了一个静态的Run方法，该方法会将指定的Action对象封装成Job，并调用Start方法利用默认的调度器进行调度。
public class Job { private readonly Action _work; public Job(Action work)=> _work = work; public JobStatus Status { get; internal set; } internal protected virtual void Invoke() { Status = JobStatus.Running; _work(); Status = JobStatus.Completed; } public void Start(JobScheduler? scheduler = null) => (scheduler ?? JobScheduler.Current).QueueJob(this); public static Job Run(Action work) { var job = new Job(work); job.Start(); return job; } }三、基于线程池的调度
Task如何执行取决于选择怎样的调度器，.NET默认采用基于线程池的调度策略，这一策略体现在ThreadPoolTaskScheduler类型上，我们使用如下这个ThreadPoolJobScheduler 进行模拟。如下面的代码片段所示，重写的QueueJob方法通过调用ThreadPool.QueueUserWorkItem方法执行指定Job对象封装的Action委托。JobScheduler的Current属性设置的默认调度器就是这么一个ThreadPoolJobScheduler 对象。
public class ThreadPoolJobScheduler : JobScheduler { public override void QueueJob(Job job) { job.Status = JobStatus.Scheduled; var executionContext = ExecutionContext.Capture(); ThreadPool.QueueUserWorkItem(_ => ExecutionContext.Run(executionContext!, _ => job.Invoke(), null)); } }
我们按照如下的方式调用Job的静态Run方法创建并执行了三个Job，每个Job封装的Action委托在执行的时候会将当前线程ID打印出来。
_ = Job.Run(() => Console.WriteLine($"Job1 is excuted in thread {Thread.CurrentThread.ManagedThreadId}")); _ = Job.Run(() => Console.WriteLine($"Job2 is excuted in thread {Thread.CurrentThread.ManagedThreadId}")); _ = Job.Run(() => Console.WriteLine($"Job3 is excuted in thread {Thread.CurrentThread.ManagedThreadId}")); Console.ReadLine();
由于采用默认的基于线程池的调度策略，所以三个Job会在三个不同的线程上执行。
我们知道.NET进程只有一个全局的线程池，对于一些需要长时间运行且具有较高优先级的操作，采用基于线程池的调用未必是好的选择。比如在一个Web应用中，线程池的工作线程会被用来处理请求，对于一个需要持续运行的Job可能会因为可用工作线程的不足而被阻塞。.NET对于这种情况具有不同的处理方式（启动Task的时候选择TaskCreationOptions.LongRunning选项），这里我们使用自定义调度器的方式来解决这个问题。如下这个DedicatedThreadJobScheduler 利用创建的“专有线程”来保证被调用的Job能够“立即”执行。线程的数量通过构造函数的参数指定，线程在无事可做的时候被“挂起”以及有新的Job被调度时被“复苏”通过一个ManualResetEvent对象来完成。
public class DedicatedThreadJobScheduler : JobScheduler { private readonly Queue<Job>[] _queues; private readonly Thread[] _threads; private readonly ManualResetEvent[] _events; public DedicatedThreadJobScheduler (int threadCount) { _queues = new Queue<Job>[threadCount]; _threads = new Thread[threadCount]; _events = new ManualResetEvent[threadCount]; for (int index = 0; index < threadCount; index++) { var queue = _queues[index] = new Queue<Job>(); var thread = _threads[index] = new Thread(Invoke); _events[index] = new ManualResetEvent(true); thread.Start(index); } void Invoke(object? state) { var index = (int)state!; var @event = _events[index]; while (true) { if (@event.WaitOne()) { while (true) { if (!_queues[index].TryDequeue(out var job)) { Suspend(index); break; } job.Invoke(); } } } } } public override void QueueJob(Job job) { job.Status = JobStatus.Scheduled; var (queue, index) = _queues.Select((queue, index) => (queue, index)).OrderBy(it => it.queue.Count).First(); queue.Enqueue(job); Resume(index); } public void Suspend(int index) => _events[index].Reset(); public void Resume(int index) => _events[index].Set(); }
还是上面演示的程序，这次我们将当前调度器设置为上面这个DedicatedThreadJobScheduler ，并将使用的线程数设置为2。
JobScheduler.Current = new DedicatedThreadJobScheduler (2); _ = Job.Run(() => Console.WriteLine($"Job1 is excuted in thread {Thread.CurrentThread.ManagedThreadId}")); _ = Job.Run(() => Console.WriteLine($"Job2 is excuted in thread {Thread.CurrentThread.ManagedThreadId}")); _ = Job.Run(() => Console.WriteLine($"Job3 is excuted in thread {Thread.CurrentThread.ManagedThreadId}")); _ = Job.Run(() => Console.WriteLine($"Job4 is excuted in thread {Thread.CurrentThread.ManagedThreadId}")); _ = Job.Run(() => Console.WriteLine($"Job5 is excuted in thread {Thread.CurrentThread.ManagedThreadId}")); _ = Job.Run(() => Console.WriteLine($"Job6 is excuted in thread {Thread.CurrentThread.ManagedThreadId}")); Console.ReadLine();
我们会发现所有的操作只会在两个固定的线程中被执行。
如果需要在某个Task执行之后接着执行后续的操作，我们可以调用其ContinueWith方法指定待执行的操作，现在我们将这个方法定义Job类型上。Job与Task的ContinueWith有些差异，在这里我们认为ContinueWith指定的也是一个Job，那么多个Job则可以按照预先编排的顺序构成一个链表。当前Job执行后，只需要将后续这个Job交付给调度器就可以了。如下面的代码片段所示，我们利用_continue字段来表示异步等待执行的Job，并利用它维持一个Job链表。ContinueWith方法会将指定的Action委托封装成Job并添加到链表末端。
public class Job { private readonly Action _work; private Job? _continue; public Job(Action work) => _work = work; public JobStatus Status { get; internal set; } public void Start(JobScheduler? scheduler = null) => (scheduler ?? JobScheduler.Current).QueueJob(this); internal protected virtual void Invoke() { Status = JobStatus.Running; _work(); Status = JobStatus.Completed; _continue?.Start(); } public static Job Run(Action work) { var job = new Job(work); job.Start(); return job; } public Job ContinueWith(Action<Job> continuation) { if (_continue == null) { var job = new Job(() => continuation(this)); _continue = job; } else { _continue.ContinueWith(continuation); } return this; } }
利用ContinueWith方法实现异步操作的按序执行体现在如下的程序中。
Job.Run(() =>{ Thread.Sleep(1000); Console.WriteLine("Foo1"); }).ContinueWith(_ =>{ Thread.Sleep(100); Console.WriteLine("Bar1"); }).ContinueWith(_ =>{ Thread.Sleep(100); Console.WriteLine("Baz1"); });
Job.Run(() =>{ Thread.Sleep(100); Console.WriteLine("Foo2"); }).ContinueWith(_ =>{ Thread.Sleep(10); Console.WriteLine("Bar2"); }).ContinueWith(_ =>{ Thread.Sleep(10); Console.WriteLine("Baz2"); });
Console.ReadLine();
输出结果
虽然ContinueWith方法能够解决“异步等待”的问题，但是我们更喜欢使用await关键字，接下来我们就为Job赋予这个能力。为此我们定义了如下这个实现了ICriticalNotifyCompletion接口的JobAwaiter结构体。顾名思义，该接口用来发送操作完成的通知。一个JobAwaiter对象由一个Job对象构建而成，当它自身执行完成之后，OnCompleted方法会被调用，我们利用它执行后续的操作。
public struct JobAwaiter: ICriticalNotifyCompletion { private readonly Job _job; public bool IsCompleted => _job.Status == JobStatus.Completed; public JobAwaiter(Job job) { _job = job; if (job.Status == JobStatus.Created) { job.Start(); } } public void OnCompleted(Action continuation) { _job.ContinueWith(_ => continuation()); } public void GetResult() { } public void UnsafeOnCompleted(Action continuation)=>OnCompleted(continuation); }
我们在Job类型上添加这个GetAwaiter方法返回根据自身创建的JobAwaiter对象。
public class Job { private readonly Action _work; private Job? _continue; public Job(Action work) => _work = work; public JobStatus Status { get; internal set; } public void Start(JobScheduler? scheduler = null) => (scheduler ?? JobScheduler.Current).QueueJob(this); internal protected virtual void Invoke() { Status = JobStatus.Running; _work(); Status = JobStatus.Completed; _continue?.Start(); } public static Job Run(Action work) { var job = new Job(work); job.Start(); return job; } public Job ContinueWith(Action<Job> continuation) { if (_continue == null) { var job = new Job(() => continuation(this)); _continue = job; } else { _continue.ContinueWith(continuation); } return this; } public JobAwaiter GetAwaiter() => new(this); }
任何一个类型一旦拥有了这样一个GetAwaiter方法，我们就能将await关键词应用在对应的对象上面。
await Foo(); await Bar(); await Baz();
Console.ReadLine(); static Job Foo() => new Job(() => { Thread.Sleep(1000); Console.WriteLine("Foo"); }); static Job Bar() => new Job(() => { Thread.Sleep(100); Console.WriteLine("Bar"); }); static Job Baz() => new Job(() => { Thread.Sleep(10); Console.WriteLine("Baz"); });
输出结果：
我想你应该知道await关键字仅仅是编译器提供的语法糖，编译后的代码会利用一个“状态机”实现“异步等待”的功能，上面这段代码最终编译成如下的形式。值得一提的是，Debug和Release模式编译出来的代码是不同的，下面给出的是Release模式下的编译结果，上述的状态机体现为生成的<<Main>$>d__0这个结构体。它的实现其实很简单：如果个方法出现了N个await关键字，它们相当于将整个方法的执行流程切割成N+1段，状态机的状态体现为当前应该执行那段，具体的执行体现在MoveNext方法上。GetAwaiter方法返回的ICriticalNotifyCompletion对象用来确定当前操作是否结束，如果结束则可以直接指定后续操作，否则需要调用AwaitUnsafeOnCompleted对后续操作进行处理。
// Program using System; using System.Diagnostics; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; using System.Threading.Tasks; using Jobs; [CompilerGenerated] internal class Program { [StructLayout(LayoutKind.Auto)] [CompilerGenerated] private struct <<Main>$>d__0 : IAsyncStateMachine { public int <>1__state; public AsyncTaskMethodBuilder <>t__builder; private JobAwaiter <>u__1; private void MoveNext() { int num = <>1__state; try { JobAwaiter awaiter; switch (num) { default: awaiter = <<Main>$>g__Foo|0_0().GetAwaiter(); if (!awaiter.IsCompleted) { num = (<>1__state = 0); <>u__1 = awaiter; <>t__builder.AwaitUnsafeOnCompleted(ref awaiter, ref this); return; } goto IL_006c; case 0: awaiter = <>u__1; <>u__1 = default(JobAwaiter); num = (<>1__state = -1); goto IL_006c; case 1: awaiter = <>u__1; <>u__1 = default(JobAwaiter); num = (<>1__state = -1); goto IL_00c6; case 2: { awaiter = <>u__1; <>u__1 = default(JobAwaiter); num = (<>1__state = -1); break; } IL_00c6: awaiter.GetResult(); awaiter = <<Main>$>g__Baz|0_2().GetAwaiter(); if (!awaiter.IsCompleted) { num = (<>1__state = 2); <>u__1 = awaiter; <>t__builder.AwaitUnsafeOnCompleted(ref awaiter, ref this); return; } break; IL_006c: awaiter.GetResult(); awaiter = <<Main>$>g__Bar|0_1().GetAwaiter(); if (!awaiter.IsCompleted) { num = (<>1__state = 1); <>u__1 = awaiter; <>t__builder.AwaitUnsafeOnCompleted(ref awaiter, ref this); return; } goto IL_00c6; } awaiter.GetResult(); Console.ReadLine(); } catch (Exception exception) { <>1__state = -2; <>t__builder.SetException(exception); return; } <>1__state = -2; <>t__builder.SetResult(); } void IAsyncStateMachine.MoveNext() { //ILSpy generated this explicit interface implementation from .override directive in MoveNext this.MoveNext(); } [DebuggerHidden] private void SetStateMachine([System.Runtime.CompilerServices.Nullable(1)] IAsyncStateMachine stateMachine) { <>t__builder.SetStateMachine(stateMachine); } void IAsyncStateMachine.SetStateMachine([System.Runtime.CompilerServices.Nullable(1)] IAsyncStateMachine stateMachine) { //ILSpy generated this explicit interface implementation from .override directive in SetStateMachine this.SetStateMachine(stateMachine); } } [AsyncStateMachine(typeof(<<Main>$>d__0))] private static Task <Main>$(string[] args) { <<Main>$>d__0 stateMachine = default(<<Main>$>d__0); stateMachine.<>t__builder = AsyncTaskMethodBuilder.Create(); stateMachine.<>1__state = -1; stateMachine.<>t__builder.Start(ref stateMachine); return stateMachine.<>t__builder.Task; } [SpecialName] private static void <Main>(string[] args) { <Main>$(args).GetAwaiter().GetResult(); } }
上面提到过，编译器生成的状态机代码在Debug和Release模式是不一样的。在Release模式下状态机是一个结构体，虽然是以接口ICriticalNotifyCompletion的方式使用它，但是由于使用了ref关键字，所以不会涉及装箱，所以不会对GC造成任何影响。但是Debug模式下生成的状态机则是一个类（如下所示），将会涉及针对堆内存的分配和回收。对于遍布await关键字的应用程序，两者之间的性能差异肯定是不同的。实际上针对Task的很多优化策略，比如使用ValueTask，对某些Task<T>对象（比如状态为Completed的Task<bool>对象）的复用，以及使用IValueTaskSource等，都是为了解决内存分配的问题。
// Program using System; using System.Diagnostics; using System.Runtime.CompilerServices; using System.Threading.Tasks; using Jobs; [CompilerGenerated] internal class Program { [CompilerGenerated] private sealed class <<Main>$>d__0 : IAsyncStateMachine { public int <>1__state; public AsyncTaskMethodBuilder <>t__builder; public string[] args; private JobAwaiter <>u__1; private void MoveNext() { int num = <>1__state; try { JobAwaiter awaiter3; JobAwaiter awaiter2; JobAwaiter awaiter; switch (num) { default: awaiter3 = <<Main>$>g__Foo|0_0().GetAwaiter(); if (!awaiter3.IsCompleted) { num = (<>1__state = 0); <>u__1 = awaiter3; <<Main>$>d__0 stateMachine = this; <>t__builder.AwaitUnsafeOnCompleted(ref awaiter3, ref stateMachine); return; } goto IL_007e; case 0: awaiter3 = <>u__1; <>u__1 = default(JobAwaiter); num = (<>1__state = -1); goto IL_007e; case 1: awaiter2 = <>u__1; <>u__1 = default(JobAwaiter); num = (<>1__state = -1); goto IL_00dd; case 2: { awaiter = <>u__1; <>u__1 = default(JobAwaiter); num = (<>1__state = -1); break; } IL_00dd: awaiter2.GetResult(); awaiter = <<Main>$>g__Baz|0_2().GetAwaiter(); if (!awaiter.IsCompleted) { num = (<>1__state = 2); <>u__1 = awaiter; <<Main>$>d__0 stateMachine = this; <>t__builder.AwaitUnsafeOnCompleted(ref awaiter, ref stateMachine); return; } break; IL_007e: awaiter3.GetResult(); awaiter2 = <<Main>$>g__Bar|0_1().GetAwaiter(); if (!awaiter2.IsCompleted) { num = (<>1__state = 1); <>u__1 = awaiter2; <<Main>$>d__0 stateMachine = this; <>t__builder.AwaitUnsafeOnCompleted(ref awaiter2, ref stateMachine); return; } goto IL_00dd; } awaiter.GetResult(); Console.ReadLine(); } catch (Exception exception) { <>1__state = -2; <>t__builder.SetException(exception); return; } <>1__state = -2; <>t__builder.SetResult(); } void IAsyncStateMachine.MoveNext() { //ILSpy generated this explicit interface implementation from .override directive in MoveNext this.MoveNext(); } [DebuggerHidden] private void SetStateMachine([System.Runtime.CompilerServices.Nullable(1)] IAsyncStateMachine stateMachine) { } void IAsyncStateMachine.SetStateMachine([System.Runtime.CompilerServices.Nullable(1)] IAsyncStateMachine stateMachine) { //ILSpy generated this explicit interface implementation from .override directive in SetStateMachine this.SetStateMachine(stateMachine); } } [AsyncStateMachine(typeof(<<Main>$>d__0))] [DebuggerStepThrough] private static Task <Main>$(string[] args) { <<Main>$>d__0 stateMachine = new <<Main>$>d__0(); stateMachine.<>t__builder = AsyncTaskMethodBuilder.Create(); stateMachine.args = args; stateMachine.<>1__state = -1; stateMachine.<>t__builder.Start(ref stateMachine); return stateMachine.<>t__builder.Task; } [SpecialName] [DebuggerStepThrough] private static void <Main>(string[] args) { <Main>$(args).GetAwaiter().GetResult(); } }