ToB企服应用市场:ToB评测及商务社交产业平台

标题: C# 异步编程async/await [打印本页]
作者: 饭宝    时间: 2023-3-10 22:01
标题: C# 异步编程async/await
概述

异步这个概念在不同语境下有不同的解释，比如在一个单核CPU里开启两个线程执行两个函数，通常认为这种调用是异步的，但对于CPU来说它是单核不可能同时运行两个函数，不过是由系统调度在不同的时间分片中执行。一般来说，如果两个工作能同时进行，就认为是异步的。在编程中，它通常代表函数的调用可以在不执行完的情况下返回，必要时在完成时回调。
有一个概念常常被混淆，多线程和异步。很多人认为异步就是多线程的，但是多线程只是实现异步的其中一种方式，除此之外还有系统中断，定时器，甚至可以自己写一个状态机实现异步（C# 的异步实现类似状态机）。
不同的编程语言有不同异步编程方法，在C#语言中，常常使用async/await等关键字，和Task等类来实现异步编程。
C#异步编程用法
  1. class Program
  2. {
  3.     static void Main(string[] args)
  4.     {
  5.         var task =  IntTask();
  6.                 Console.WriteLine("等待中...");
  7.         Console.WriteLine($"算完了？ 让我康康！ result = {task.Result}");
  8.     }
  10.     static async Task<int> IntTask()
  11.     {
  12.         Console.WriteLine("等3秒吧");
  13.         await Task.Delay(3000);
  14.         return 1;
  15.     }
  16. }
复制代码
Main函数异步调用IntTask，打印"等三秒吧",随后返回到Main函数打印“等待中”，在task.Result取值时阻塞，三秒后IntTask返回（此时Task.Result被赋值）打印“result = 1”。看一下用法：
上述只是一个极简的用法，忽略了大量的细节，可以建立一个初步的印象。
async/await和Task简介

async

用async修饰一个方法，表明这个方法可以异步执行，其返回值必须是void/Task/Task（T是返回值类型）其中一个，方法内的语句至少包含一个await关键字，否则会被同步的方式执行。
await

await只能修饰（返回值是）Task类型变量，此时会返回Task.Result或void而不是Task本身，在上述示例中，Main没有被async修饰，不能使用await，其返回值就是Task, 而IntTask调用Task.Delay就是直接返回void。await也只能在被async修饰的函数的语句中使用。
Task

源于基于任务的异步模式(Task-based Asynchronous Pattern,TAP)，被作为异步函数的返回值。异步函数的返回值有三种：
其他

实现原理剖析

如果使用反汇编等手段，可以看到上述示例代码的编译：

在返回1之前，好像有什么“奇怪的东西”被调用，编译器又背着开发者偷偷干了什么呢？
实现原理示例

在微软的开发博客里有一个叫谢尔盖·杰普利亚科夫（Sergey Tepliakov）的毛子曾提到这部分，来看一下他的示例：
源码
  1. class StockPrices
  2. {
  3.     private Dictionary<string, decimal> _stockPrices;
  4.     public async Task<decimal> GetStockPriceForAsync(string companyId)
  5.     {
  6.         await InitializeMapIfNeededAsync();
  7.         _stockPrices.TryGetValue(companyId, out var result);
  8.         return result;
  9.     }
  11.     private async Task InitializeMapIfNeededAsync()
  12.     {
  13.         if (_stockPrices != null)
  14.             return;
  16.         await Task.Delay(42);
  17.         // Getting the stock prices from the external source and cache in memory.
  18.         _stockPrices = new Dictionary<string, decimal> { { "MSFT", 42 } };
  19.     }
  20. }
复制代码
这是他的源代码，这个类叫做StockPrices（股票价格），其核心业务是根据公司ID查询股票价格GetStockPriceForAsync，这是一个异步调用，首先它先异步调用InitializeMapIfNeededAsync对数据库进行初始化，初始化完成尝试从数据库中获取该公司的股票价格返回。
上述提到编译器偷偷自己生成了代码，如果手动实现大概是怎样的呢？来看谢尔盖给出的解：
手动实现
  1. class GetStockPriceForAsync_StateMachine
  2. {
  3.     enum State { Start, Step1, }
  4.     private readonly StockPrices @this;
  5.     private readonly string _companyId;
  6.     private readonly TaskCompletionSource<decimal> _tcs;
  7.     private Task _initializeMapIfNeededTask;
  8.     private State _state = State.Start;
  10.     public GetStockPriceForAsync_StateMachine(StockPrices @this, string companyId)
  11.     {
  12.         this.@this = @this;
  13.         _companyId = companyId;
  14.     }
  16.     public void Start()
  17.     {
  18.         try
  19.         {
  20.             if (_state == State.Start)
  21.             {
  22.                 // The code from the start of the method to the first 'await'.
  24.                 if (string.IsNullOrEmpty(_companyId))
  25.                     throw new ArgumentNullException();
  27.                 _initializeMapIfNeededTask = @this.InitializeMapIfNeeded();
  29.                 // Update state and schedule continuation
  30.                 _state = State.Step1;
  31.                 _initializeMapIfNeededTask.ContinueWith(_ => Start());
  32.             }
  33.             else if (_state == State.Step1)
  34.             {
  35.                 // Need to check the error and the cancel case first
  36.                 if (_initializeMapIfNeededTask.Status == TaskStatus.Canceled)
  37.                     _tcs.SetCanceled();
  38.                 else if (_initializeMapIfNeededTask.Status == TaskStatus.Faulted)
  39.                     _tcs.SetException(_initializeMapIfNeededTask.Exception.InnerException);
  40.                 else
  41.                 {
  42.                     // The code between first await and the rest of the method
  44.                     @this._store.TryGetValue(_companyId, out var result);
  45.                     _tcs.SetResult(result);
  46.                 }
  47.             }
  48.         }
  49.         catch (Exception e)
  50.         {
  51.             _tcs.SetException(e);
  52.         }
  53.     }
  55.     public Task<decimal> Task => _tcs.Task;
  56. }
  58. public Task<decimal> GetStockPriceForAsync(string companyId)
  59. {
  60.     var stateMachine = new GetStockPriceForAsync_StateMachine(this, companyId);
  61.     stateMachine.Start();
  62.     return stateMachine.Task;
  63. }
复制代码
从类名GetStockPriceForAsync_StateMachine可以看到，他为这个异步调用生成了一个状态机来实现异步，先来看下成员变量：
现在看来这段代码的逻辑就比较清楚了，在调用异步查询股票的接口时，创建了一个状态机并调用状态机的Start函数，第一次进入start函数时状态机的状态是Start状态，它给_initializeMapIfNeededTask赋值，把状态机状态流转到Step1，并让_initializeMapIfNeededTask执行结束末尾再次调用Start函数（ContinueWith）。
_initializeMapIfNeededTask任务在等待了42毫秒后(Task.Delay(42))，末尾时再次调用了Start函数，此时状态为Step1。首先检查了Task状态，符合要求调用_tcs.SetResult（其实是给Task的Result赋值），此时异步任务完成。
TaskCompletionSource

看官方文档给的定义：
表示未绑定到委托的 Task 的制造者方，并通过 Task 属性提供对使用者方的访问
简单的示例：
  1. static void Main()
  2. {
  3.         TaskCompletionSource<int> tcs1 = new TaskCompletionSource<int>();
  4.         Task<int> t1 = tcs1.Task;
  6.         // Start a background task that will complete tcs1.Task
  7.         Task.Factory.StartNew(() =>
  8.         {
  9.                 Thread.Sleep(1000);
  10.                 tcs1.SetResult(15);
  11.         });
  12. }
复制代码
看的出来这个类就是对Task的包装，方便创建分发给使用者的任务。其核心就是包装Task并方便外面设置其属性和状态
Task.ContinueWith

创建一个在目标 Task 完成时异步执行的延续任务
可以传入一个委托，在Task完成的末尾调用。这是一个典型的续体传递风格（continuation-pass style）。
续体传递风格

续体传递风格（continuation-pass style, CPS），来看维基百科的描述:
A function written in continuation-passing style takes an extra argument: an explicit "continuation"; i.e., a function of one argument. When the CPS function has computed its result value, it "returns" it by calling the continuation function with this value as the argument. That means that when invoking a CPS function, the calling function is required to supply a procedure to be invoked with the subroutine's "return" value. Expressing code in this form makes a number of things explicit which are implicit in direct style. These include: procedure returns, which become apparent as calls to a continuation; intermediate values, which are all given names; order of argument evaluation, which is made explicit; and tail calls, which simply call a procedure with the same continuation, unmodified, that was passed to the caller
大概的意思是，这种风格的函数比起普通的有一个额外的函数指针参数，调用结束（即将return）调用或者函数参数（替代直接return到调用者Caller）。还有一些其他细节，就不多说了，感兴趣自行翻译查看。
来看一个极简的例子：
int a = b + c + d;
这是一个链式运算，是有顺序的,在C++中，上述运算其实是：
int a = (b + c) + d; 先计算tmp = b + c（tmp是寄存器上一个临时的值，也称将亡值），然后计算 int a = tmp + c
使用续体传递来模拟这一过程：
  1. class Program
  2. {
  3.     public class Result<T>
  4.     {
  5.         public T V;
  6.     }
  8.     static void Main(string[] args)
  9.     {
  10.         int a = 1;
  11.         int b = 2;
  12.         int c = 3;
  13.         int d = 4;
  15.         Result<int> ar = new() { V = a};
  16.         Calc3(ar, b, c, d, Calc2);
  18.         Console.WriteLine($"a = {ar.V}");
  19.     }
  21.     static void Calc3(Result<int> ar, int b, int c, int d, Action<Result<int>, int, int> continues)
  22.     {
  23.         int tmp = b + c;
  24.         continues(ar, tmp, d);
  25.     }
  27.     static void Calc2(Result<int> ar, int tmp, int d)
  28.     {
  29.         ar.V = tmp + d;
  30.     }
  31. }
复制代码
上述代码应该很清楚了，稍微看下应该能看明白。
C#编译器的实现
  1. struct _GetStockPriceForAsync_d__1 : IAsyncStateMachine
  2. {
  3.     public StockPrices __this;
  4.     public string companyId;
  5.     public AsyncTaskMethodBuilder<decimal> __builder;
  6.     public int __state;
  7.     private TaskAwaiter __task1Awaiter;
  9.     public void MoveNext()
  10.     {
  11.         decimal result;
  12.         try
  13.         {
  14.             TaskAwaiter awaiter;
  15.             if (__state != 0)
  16.             {
  17.                 // State 1 of the generated state machine:
  18.                 if (string.IsNullOrEmpty(companyId))
  19.                     throw new ArgumentNullException();
  21.                 awaiter = __this.InitializeLocalStoreIfNeededAsync().GetAwaiter();
  23.                 // Hot path optimization: if the task is completed,
  24.                 // the state machine automatically moves to the next step
  25.                 if (!awaiter.IsCompleted)
  26.                 {
  27.                     __state = 0;
  28.                     __task1Awaiter = awaiter;
  30.                     // The following call will eventually cause boxing of the state machine.
  31.                     __builder.AwaitUnsafeOnCompleted(ref awaiter, ref this);
  32.                     return;
  33.                 }
  34.             }
  35.             else
  36.             {
  37.                 awaiter = __task1Awaiter;
  38.                 __task1Awaiter = default(TaskAwaiter);
  39.                 __state = -1;
  40.             }
  42.             // GetResult returns void, but it'll throw if the awaited task failed.
  43.             // This exception is catched later and changes the resulting task.
  44.             awaiter.GetResult();
  45.             __this._stocks.TryGetValue(companyId, out result);
  46.         }
  47.         catch (Exception exception)
  48.         {
  49.             // Final state: failure
  50.             __state = -2;
  51.             __builder.SetException(exception);
  52.             return;
  53.         }
  55.         // Final state: success
  56.         __state = -2;
  57.         __builder.SetResult(result);
  58.     }
  60.     void IAsyncStateMachine.SetStateMachine(IAsyncStateMachine stateMachine)
  61.     {
  62.         __builder.SetStateMachine(stateMachine);
  63.     }
  64. }
  66. [AsyncStateMachine(typeof(_GetStockPriceForAsync_d__1))]
  67. public Task<decimal> GetStockPriceFor(string companyId)
  68. {
  69.     _GetStockPriceForAsync_d__1 _GetStockPriceFor_d__;
  70.     _GetStockPriceFor_d__.__this = this;
  71.     _GetStockPriceFor_d__.companyId = companyId;
  72.     _GetStockPriceFor_d__.__builder = AsyncTaskMethodBuilder<decimal>.Create();
  73.     _GetStockPriceFor_d__.__state = -1;
  74.     var __t__builder = _GetStockPriceFor_d__.__builder;
  75.     __t__builder.Start<_GetStockPriceForAsync_d__1>(ref _GetStockPriceFor_d__);
  76.     return _GetStockPriceFor_d__.__builder.Task;
  77. }
复制代码
比较一下C#编译器生成的状态机：
上述成员有一些和之前手撸的状态机不太一样，等下面会介绍，先来这一套的逻辑：
首先创建了一个_GetStockPriceForAsync_d__1状态机_GetStockPriceFor_d__并初始化赋值，随后调用了这个状态机的__builder的Start函数并把该状态机作为引用参数传入。__builder.Start函数会调到该状态机的MoveNext函数（下面会介绍），这和手撸代码状态机Start函数调用类似。
MoveNext与Start函数的处理过程也类似：第一次进来__state == -1，__builder.AwaitUnsafeOnCompleted切换上下文执行InitializeLocalStoreIfNeededAsync异步任务，并指定在完成后切换到当前上下文调用该状态机的MoveNext函数，类似手撸代码的Task.ContinueWith。第二次进入时，执行到__builder.SetResult(result)，异步任务基本完成。
上述描述也是忽略了一些细节，下面是调用的时序图，会更清楚些，有些不太清楚的点后面会详细介绍。

TaskAwaiter

来看下官方定义：
public readonly struct TaskAwaiter : System.Runtime.CompilerServices.ICriticalNotifyCompletion 提供等待异步任务完成的对象
结构：

可以看到，这个所谓“等待异步任务完成的对象”，主要是保证实现ICriticalNotifyCompletion的接口OnCompleted等。
AsyncTaskMethodBuilder(TAwaiter, TStateMachine)

官方定义：

个人认为可以视为异步任务的“门面”，它负责启动状态机，传递一些中间状态，并在最终SetResult时表示它和其子例程的异步任务结束。其中有一个方法AwaitUnsafeOnCompleted，值得研究一下。
AsyncTaskMethodBuilder.AwaitUnsafeOnCompleted

这个方法在上述中一笔带过，被描述为类似Task.ContinueWith，确实如此，但执行过程相当复杂，在这里也只是简单介绍下过程。
AwaitUnsafeOnCompleted首先会调用GetCompletionAction，GetCompletionAction创建了一个保存了上下文 context = ExecuteContext.FastCapture()的MoveNextRunner，并返回了指向的MoveNextRunner.Run函数的委托。
接着调用参数awaiter的UnsafeOnCompleted（completionAction）函数，这里completionAction就是上述的那个委托，内部调用了成员Task.SetContinuationForAwait函数来初始化续体，SetContinuationForAwait又调用AddTaskContinuation把延续方法添加到Task中，当上述示例源码中的InitializeMapIfNeededAsync函数执行完调用Runner.Run：
  1. [SecuritySafeCritical]
  2. internal void Run()
  3. {
  4.     if (this.m_context != null)
  5.     {
  6.         try
  7.         {
  8.             // 我们并未给 s_invokeMoveNext 赋值，所以 callback == null
  9.             ContextCallback callback = s_invokeMoveNext;
  10.             if (callback == null)
  11.             {
  12.                 // 将回调设置为下方的 InvokeMoveNext 方法
  13.                 s_invokeMoveNext = callback = new
  14.                 ContextCallback(AsyncMethodBuilderCore.MoveNextRunner.InvokeMoveNext);
  15.             }
  16.             ExecutionContext.Run(this.m_context, callback, this.m_stateMachine, true);
  17.             return;
  18.         }
  19.         finally
  20.         {
  21.             this.m_context.Dispose();
  22.         }
  23.     }
  24.     this.m_stateMachine.MoveNext();
  25. }
  27. [SecurityCritical]
  28. private static void InvokeMoveNext(object stateMachine)
  29. {
  30.     ((IAsyncStateMachine) stateMachine).MoveNext();
  31. }
复制代码
((IAsyncStateMachine) stateMachine).MoveNext() 重新调用了状态机的MoveNext()
参考链接


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4