个人对 C# 中 async/await 的理解

如题,不一定对,欢迎佬友们一起交流(纯水文,进来的都给我点赞!!!)

主要参考的下面两篇文章:

块引用中的内容是我觉得有意思(可以水)加上的,可能和主题无关,追求连贯的话(写完发现已经不连贯了)看的时候可以跳过

讲讲我理解的 Task

Task,在其他语言中可能叫 Future 或者 Promise,表示一个结果的代理(可以获取结果的操作,以及操作完成产生的结果)。这里的操作可能是 I/O 操作,也可能是 CPU 密集型操作。
这里的操作需要一个线程来承载,对于 CPU 密集型操作,操作在某个线程上执行,对于 I/O 操作,操作在某个线程上等待操作系统的事件通知。

async/await 的语义和实现

首先要知道,async/await 是编译时功能,编译器会将 async/await 代码转换为状态机。

有一个将 async/await 变为运行时功能的实验

await 用于 Task,表示

  • 如果 Task 未完成,将 await 之后的代码作为回调,注册到这个未完成的 Task 上,然后返回(控制流回到调用者)
  • 如果 Task 已完成,获取结果(并且接着执行 await 之后的代码)(例如读文件,但要读的文件内容已经在内存中,不用等待磁盘 I/O。或者创建 Task 后执行了其他代码再 await,此时 Task 已经完成)

关于异步方法的传染性,我的理解是方法中 await 了一个 Task,那么这个方法就可能在没有获取到结果(也就是没有执行完整个方法)的情况下返回,那么这个方法也就成为了一个结果的代理(可以获取结果的操作,也就是 Task)。

async 告诉编译器需要为这个方法生成一个状态机。因为 await 之后的代码执行可能需要用到方法的参数和方法中的局部变量,以及保存接下来从哪里继续执行。

先看一个简单的方法:

internal class Program {
    private static async Task Main() {
        using var client = new HttpClient();
        var result = await client.GetStringAsync("https://www.bilibili.com");
        Console.WriteLine(result.Length);
    }
}

ILSpy 反编译,选择语言版本 C# 1.0
(简化部分错误处理和无关逻辑,改掉了编译器自动生成的命名)

internal class Program {
    private sealed class <Main>d__0 : IAsyncStateMachine {
        public int state;
        public AsyncTaskMethodBuilder builder;
        private HttpClient client = new HttpClient();
        private string result;
        private TaskAwaiter<string> awaiter;

        private void MoveNext() {
            try {
                if (state != 0) {
                    awaiter = client.GetStringAsync("https://www.bilibili.com").GetAwaiter();
                    if (!awaiter.IsCompleted) {
                        state = 0;
                        builder.AwaitUnsafeOnCompleted(ref awaiter, ref this);
                        return;
                    }
                }
                result = awaiter.GetResult();
                Console.WriteLine(result.Length);
            } catch (Exception exception) {
                state = -2;
                builder.SetException(exception);
                return;
            }
            state = -2;
            builder.SetResult();
        }
    }

    [AsyncStateMachine(typeof(<Main>d__0))]
    private static Task Main() {
        <Main>d__0 stateMachine = new <Main>d__0();
        stateMachine.builder = AsyncTaskMethodBuilder.Create();
        stateMachine.state = -1;
        stateMachine.builder.Start(ref stateMachine);
        return stateMachine.builder.Task;
    }

    private static void <Main>() {
        Main().GetAwaiter().GetResult();
    }
}

可以看到,编译器为原来的 Main 方法生成了一个状态机类 <Main>d__0。执行现在的 Main 方法时,会调用 Start 方法。这个方法会调用状态机的 MoveNext 方法。

awaiter = client.GetStringAsync("https://www.bilibili.com").GetAwaiter(); 可以看作创建了一个返回 string 的 Task。

如果这个 Task 未完成(awaiter.IsCompleted 为 false),那么就保存下一次从哪里开始执行(state = 0),并且注册自身作为回调(builder.AwaitUnsafeOnCompleted(ref awaiter, ref this))。回调会在 Task 完成时被调用。
如果已完成,就直接获取结果(result = awaiter.GetResult();),并且执行后续代码。

上面的例子里,<Main> 方法中调用了 GetResult 方法等待 Task 完成,而不是注册回调。因为 <Main> 方法是主线程的入口方法,主线程结束时进程就结束了,所以必须等待 Main 方法的 Task 完成。

这里的 GetAwaiter 是鸭子类型的一个应用,只要一个类型实现了 GetAwaiter 方法,并且返回的对象实现了 IsCompleted 属性以及 GetResult 和 OnCompleted/UnsafeOnCompleted 方法,就可以 await 这个类型的对象。 这里 Task 实现了 GetAwaiter 方法,返回 TaskAwaiter 类型的对象

类似于一个类型如果实现了 GetEnumerator 方法,且返回的对象实现了 Current 属性以及 MoveNext 方法,就可以用 foreach 遍历这个类型的对象。

实际上类似的不止这些,在返回 IEnumerable<T> 的方法中使用 yield return,编译器会为这个方法生成一个状态机类,实际上用状态机实现 async/await 就是由迭代器启发的,具体可以看 How Async/Await Really Works in C# - .NET Blog 这篇

public static IEnumerable<int> Fib() {
    int prev = 0, next = 1;
    yield return prev;
    yield return next;

    while (true) {
        int sum = prev + next;
        yield return sum;
        prev = next;
        next = sum;
    }
}

ILSpy 反编译,选择语言版本 C# 1.0,会生成类似下面的代码

private sealed class <Fib>d__1 : IEnumerable<int>, IEnumerable, IEnumerator<int>, IEnumerator, IDisposable {
    private int state;
    private int current;
    private int initialThreadId;
    private int prev;
    private int next;
    private int sum;

    int IEnumerator<int>.Current => current

    object IEnumerator.Current => current

    public <Fib>d__1(int state) {
        this.state = state;
        initialThreadId = Environment.CurrentManagedThreadId;
    }

    void IDisposable.Dispose() {
        state = -2;
    }

    private bool MoveNext() {
        switch (state) {
        default:
            return false;
        case 0:
            state = -1;
            prev = 0;
            next = 1;
            current = prev;
            state = 1;
            return true;
        case 1:
            state = -1;
            current = next;
            state = 2;
            return true;
        case 2:
            state = -1;
            break;
        case 3:
            state = -1;
            prev = next;
            next = sum;
            break;
        }
        sum = prev + next;
        current = sum;
        state = 3;
        return true;
    }

    bool IEnumerator.MoveNext() {
        return this.MoveNext();
    }

    void IEnumerator.Reset() {
        throw new NotSupportedException();
    }

    IEnumerator<int> IEnumerable<int>.GetEnumerator() {
        if (state == -2 && initialThreadId == Environment.CurrentManagedThreadId) {
            state = 0;
            return this;
        }
        return new <Fib>d__1(0);
    }

    IEnumerator IEnumerable.GetEnumerator() {
        return ((IEnumerable<int>)this).GetEnumerator();
    }
}

[IteratorStateMachine(typeof(<Fib>d__1))]
public static IEnumerable<int> Fib() {
    return new <Fib>d__1(-2);
}

可以看到两种状态机同样可能会被多次调用,同样保存了执行状态和下次执行开始的位置

更进一步,还可以使用 IAsyncEnumerable<T> 作为迭代器的返回类型。使用 await foreach 语句对迭代器的结果进行迭代

async IAsyncEnumerable<int> GenerateNumbersAsync(int count){
    for (int i = 0; i < count; i++){
        yield return await ProduceNumberAsync(i);
    }
}

async Task<int> ProduceNumberAsync(int seed){
    await Task.Delay(1000);
    return 2 * seed;
}

有兴趣可以自己反编译看看

两种用例

CPU 密集型操作

Console.WriteLine("开始计算");
var result = await Task.Run(() => {
    // 模拟计算
    Thread.Sleep(TimeSpan.FromSeconds(1));
    return 0;
});
Console.WriteLine(result);

对于 CPU 密集型操作,我们会把耗时的方法放到 Task.Run 中执行。

Task.Run 会创建一个 Task,并且将其放进线程池的工作队列。(线程池里的线程可以简单认为做的就是循环从工作队列中取任务执行)

线程池线程被创建后执行的代码:
Source Browser

任务执行完后,会调用注册的回调(也就是 Console.WriteLine(result);)打印出结果。

I/O 密集型操作

下面是一个模拟读文件的例子:

Console.WriteLine("模拟读文件");
var result = await EmulateReadAsync();
Console.WriteLine(result);

Task<string> EmulateReadAsync() {
    var source = new TaskCompletionSource<string>();
    _ = Task.Run(() => {
        var result = EmulateRead();
        source.SetResult(result);
    });
    return source.Task;
}

// 模拟 I/O 操作
string EmulateRead() {
    // 等待系统调用返回
    Thread.Sleep(TimeSpan.FromSeconds(1));
    return "content";
}

前面说过 I/O 操作也需要一个线程来承载,等待操作系统的事件通知。I/O 操作完成后,再调用注册的回调。

区别无非在于通知的方式(阻塞调用返回,epoll,IOCP…),回调执行的位置(当前线程,线程池的线程)

贴一张偷的异步请求流程图(Windows)

有兴趣的可以看看源码,FileStream.ReadAsync 最终会调用 RandomAccess.ReadAtOffsetAsync 方法,对应图中 2 的过程。

async/await 的优势

  • 代码可读性更好,避免了回调地狱,并且对于不用回调的情况(Task 已完成)做了优化
  • 对于一些特殊的 I/O 操作(epoll,IOCP),只需要阻塞一个线程(I/O 多路复用),提高性能
  • 对于 UI 线程,避免了阻塞 UI 线程,提高响应性能

一点碎碎念

  • 想到哪就写到哪了,可能有些地方不太连贯,而且只是大概梳理了一下,很多细节都被我省略了,欢迎指正
  • 还有些想写的地方没写到,比如 ExecutionContext,ValueTask,SynchronizationContext,TaskScheduler, 等等,后面有机会再补充吧
33 Likes

task是effect吗。。可以取消吗。。可以组合吗。。

5 Likes

可视为“effect”。可通过CancellationToken取消。可使用Task.WhenAll、Task.WhenAny等组合。

1 Like

我的意思是它能不能类似haskell的bind那样将多个task组合成一个大的task,而不用实际去执行task内部的东西~

1 Like

Task 默认立即启动,若要延迟执行,可考虑 Func<Task<T>>,带有函数委托的 async/await 可参见 C# - Async/await with a Func delegate | makolyte 。

1 Like

君亦可参见 LanguageExt 的 Aff<T>,更近 Haskell 的 Monad 语义。

佬,这就是所谓的编程代码么

是不是async/await的实现都是通过迭代器+状态机(c#和js刚好我都有接触,感觉底层很像)

不错的框架。。应该把IO Monad所有的基础工作都做好了~
谢谢

虽然不会C,但是感觉当年写出异步操作方法的大神真是天才。。。

活到老学到老

没用过怎么 js 捏,不过佬友这么说了我感觉很可能是差不多的
可能底层调度方式不同吧,C# 里默认是用线程池调度的,js 似乎是事件循环?

1 Like

非佬,一般路过佬友罢了
是滴捏,编程代码很神奇吧

async2马上就来了,已经合并进主线了

真的假的真的假的 :astonished_face:
能来个链接吗佬友

跟框架有关,桌面端是线程池,但wpf记得是事件循环,wasm上也是事件循环

1 Like

不懂函数式捏:sleeping_face:
没能让佬友尽兴果咩捏

太强了 现在都不想研究底层实现 :tieba_087:

嗯嗯
Task.Run 创建的任务会在 TaskScheduler.Default 上被调度

public static Task Run(Action action) {
    return InternalStartNew(null, action, null, default, TaskScheduler.Default,
        TaskCreationOptions.DenyChildAttach, InternalTaskOptions.None);
}

这里的 TaskScheduler.Default 类型是 ThreadPoolTaskScheduler

要选择调度器可以调用 Task.Factory.StartNew 方法(但是这个方法的默认调度器却是 TaskScheduler.Current 我觉得挺奇怪的)

await 的时候如果没有调用 ConfigureAwait(false),当前任务就会捕获 SynchronizationContext.Current,任务完成后将回调运行到这个 SynchronizationContext 上,WinForm 和 WPF 好像都有自己的 SynchronizationContext,虽然我没看过,但是应该就是把回调排队到 UI 线程上