594飞飘的个人博客

学习笔记与生活记录

深入异步通用概念

发表于 更新于 分类于 ， ，
本文字数： 9.9k 阅读时长 ≈ 18 分钟

深入异步通用概念

协程是一个简单的概念，但它同时又没那么简单，因为其所代表的异步编程的思想具有一定的历史发展过程。这些东西曾经困扰了我许久，我饱读各种相关的资料却始终不得其解，因此我写下了这篇博客

前言: 本文所说的异步为单线程情况下的异步 IO。我将试图把 IO 多路复用、异步 IO、事件循环、协程、async/await 等分散的概念串联起来，方便理解其前因后果。注意其中涉及到的 syscall 以 Linux 为主。

在异步之旅开始之前

相信任何一个朋友在一开始接触编程的时候都会编写类似下面这样的 C++ 代码

1
2
3
4
5
6
7
8
#inlcude <iostream>

int main(){
  int input;
  std::cin >> input;
  std::cout << input << std::endl; 
}

作为一个 Hello world 级别的程序，大家都知道在第五行会阻塞起来等待输入，但等待输入的行为实际上并没有那么简单，在学习了操作系统之后，我们知道在这一段代码背后是调用了阻塞的 syscall read()，当你通过标准库对这个系统调用的封装调用到它时，会从用户态陷入内核态，等待内核将输入交给程序，才会继续下去。这是程序最简单的形态，以至于如果你参加任何算法竞赛的时候，要写的程序都是这样的形态。而这种系统调用因此被称为阻塞的，阻塞IO。而在等待内核将输入的信息传递给程序之前，程序处于一个休眠状态，对于操作系统来说，也就是不把运行的时间片分配给它，这种行为我们也称为把程序挂起了。

但这太简单了，初学者一定会好奇，这样一些从上到下顺序执行的代码，如何拼凑出一个带 UI 的程序。比如初学者可能会使用 Qt 去做一个贪吃蛇程序，在当下的这个年代，各种成熟的 UI 库层出不穷，很容易就能做到这一点。在做完这些刚刚入门编程的时候，他们略深入一些便会听说，他们所做的 Qt 程序运行在一个事件循环上，而 Qt 默认居然是单个线程就可以运行的。单线程怎么可能做到这个事情呢，它怎么能做到等待你输入，等待你点击 UI 的时候还能保证软件其他部分被渲染，动画似乎是完全与你的行为无关，一点也不像在 cli 中会阻塞，这种行为在多线程的前提下简直是个魔法嘛。

是的，我当年就是这么想的。在当时的我看来，阻塞完全是一个 debuff，应该需要一个不阻塞的接口。如果没有按下，也能正常执行后面的程序。那么我们就可以说，在这种情况下，我们的程序同操作系统实际做的事情一样，那就是创建一个巨大的 while 循环，不断询问操作系统对应的按钮是否被按下，这些询问本身并不花时间，只是问一问操作系统当前这个 IO 任务是否被完成。如果完成，我程序就可以从操作系统里取，如果没完成，那我就不管了，等到 while 下一次询问。这种类型的系统调用确实存在，他们被被称为非阻塞的，非阻塞IO，就像下面这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 假设这里的isClickedButtonX()就是一个不阻塞的询问
while(1) {
  if(isClickedButtonW()) {
    // do something
  }
  if(isClickedButtonA()) {
    // do something
  }
  if(isClickedButtonS()) {
    // do something
  }
  if(isClickedButtonD()) {
    // do something
  }
}

我们可以把所有的输入，包括这边代码中的 isClickedButtonX()，以及更为统一的鼠标指针，键盘输入，网络请求，甚至是外设(蓝牙，LED)这种在冯诺依曼结构里被称为输入输出设备向系统内部传递信息并交给操作系统，进而交由应用程序的所有信息称之为 IO 或者说 IO 事件。在上述的代码中，我们进行循环查询，并合理的把它们处理：也就是注释所说的 do something，不同的事件有不同处理，处理的方式在下文被称为任务，这是所谓的事件分发。这其实是最高效的处理这些 IO 事件的方式，事实上某些单片机程序就是这么做的，这种循环查询某个事件是否就绪的模式称为轮询(polling)。

这样的方式可以进行的原因是：我们必须理解程序的运行，程序的信息来自于两处，一个是来自操作系统，一个是程序自身已经编入的信息，它在运行时的额外信息完全来自于操作系统，而操作系统的信息来自于硬件的交互。操作系统本身是异步的，因为硬件是并行处理的，操作系统通过中断以及各种调度算法处理并存储来自硬件的各种信息，本身 OS 就作为一个巨大 Buffer 为程序提供信息，所以程序本身无需在意数据是怎么被具体准备出来的，你只需要向操作系统注册需要的 IO 事件，操作系统就为你在缓冲区里准备好 IO 事件发生时拿到的数据，你自己去取，因此数据其实短期内不会丢失的，只等你的应用去响应并消费。

但是这种方式最大的问题是，这样的行为意味着你的 while 代码块中的代码会一往无前的运行，每次循环都不带停的。当裸的 while(1) 出现在你的程序里，导致死循环，会大量占用 CPU 资源，使 CPU 占用率增高。

这里需要复习一下 CPU 占用率是什么。它的定义是通过对比忙碌时间和总时间得出的百分比，而总时间减去忙碌时间是空闲时间，其实操作系统完成了对 CPU 的调度，操作系统在运行时可以告诉 CPU，你要在某一个时间周期里工作多久，然后这个时间周期里的 CPU 还能休息一会，停止工作，而这段时间就是空闲时间。

而死循环程序会导致它试图最大程度请求 CPU 去做无用的轮询，操作系统为了最大满足它索要的计算资源，不允许 CPU 停止休息，因此占用率便提高了。

所以完全阻塞等待某个资源不行，开个循环询问所有任务的就绪状态也有问题，那应该怎么办呢?

破局之法之多路复用

上文提到的阻塞 IO 并没有死循环的问题，你哪怕在一个 while 循环里写一个 std::cin 也不会出现占用超多 CPU 资源的问题，因为在你运行到的时候，系统就已经定在那了，程序被挂起了，不会走新的一轮 while 循环。但问题就在于程序此时此刻就只能做一件事情了。也许你会同早期的我一样，认为这个关键就在于 syscall 阻塞了线程的活动，应该彻底抛开可能导致阻塞的系统调用，也许存在一个可以天降信息的超级接口可以传递信息，很不幸，事实证明这样的接口并不存在。

让我们回看一下令我们困惑的例子：如果我们要做一个单线程的贪吃蛇的小程序，我们可能需要程序以一个固定的帧率去绘制，移动屏幕上的这条小蛇，那么我们就需要设定一个计时器来定时做一些渲染与计算的工作。但是我们在这个问题上往往忽略了一个事情，那就是计时器事件本身就也是一个 IO 事件，从更大的一个层次上，他和你等待一个按钮按下没有什么区别。

如果在 while 里将轮询事件能解决问题，但是没有阻塞导致占用率高，但阻塞的情况只会响应一个事件，那又不行。但如果我们能把这些个 if 扔给操作系统去考虑呢？于是 IO 多路复用出现了，它的核心原理是，在一个线程里通过有且仅有一个接口去获取所有 IO 事件，如果查询的时候存在准备好的数据，则立即返回并处理，如果查询时没有准备好的数据，则立即阻塞，等待操作系统准备好数据立即将其唤醒，这一类接口，比较常见的就是 Linux 下的 select/poll/epoll，将这样的接口置于while循环下，如果你处理 IO 事件的速度高于 IO 事件产生的速度，你就可以及时解决所有 IO 事件

看到了么，我特意强调了阻塞的字眼，为什么？需要强调的是，如我程序读写一个文件的时候，只是等在那，不会读取别的比如按钮点击事件等这样程序处理事件意义上的阻塞和接口的阻塞这两个阻塞说的并不是同样的概念。接口的阻塞是因为至少目前为止要想释放 CPU 的利用率，是必须要让用户态应用程序挂起的。所以从这个意义上来说，阻塞是必然的，所有这类要把 CPU 资源谦让出来的程序都应当是阻塞的程序。而只要有一个接口访问 IO 事件，等事件一出现，就立即响应，消费掉任务，然后挂起，就不算程序处理事件意义上的阻塞，因为你的程序仍然会高效的处理事件。

是的，此阻塞非彼阻塞，所谓阻塞亦能实现非阻塞

所以，我们的程序需要高效的处理或者说消费掉 IO 事件，这是保证任务不会堆起来，所有任务都能得到处理的前提。从具体参数上看就是处理 IO 事件的速度高于 IO 事件产生的速度，所以这个条件成立的程序，是为 IO 密集型程序，适合这种模式。而不满足这个条件的程序，一种是计算量特别大的程序，可能是执行某些科学计算，求解微分方程，处理图像，甚至是深度学习任务，这些程序是为计算密集型任务，一般选择多线程方式，还有一种是现代设备 IO 速度越来越快，此时可能直接轮询，试图达到速率上限能获得更高的收益(比如 io_uring 就是在做这里的极致优化)。

事件循环

最根本的问题解决了，接下来是要为这个简陋的 while(1) 循环加工一下了，我们需要一个在这个循环体进行一次时能够彻底消费掉这些操作系统里出现的数据的代码，其实这是一个调度器。在绝大部分的实现中，它都以一些我们刚刚提到的另一个耳熟能详的名字出现：事件驱动的模式，或者说事件循环。

来看下面这段给出的伪代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
auto *task_list = TaskList.getInstance()->add(InitTask);
//你需要在这里的InitTask注册监听的事件，否则程序立即退出
auto *event_fetcher = EventFetcher.getInstance();
while(true) {
    for(auto task: *task_list)
        task();
    task_list.clear();
    auto events = EventFetcher.getInstance()->getEvents();
    // 这里如果获取不到会阻塞等待
    for(auto event: events)
        EventDispatcher(event); 
    // 分发后，后续任务也就是回调用闭包形式传入序列
    // 倘若监听的事件和任务队列全无，则程序退出
    if(!events.length() && task_list.length())
        break;
}

首先，既然要监听事件，必须有一个监听器，在这里是 EventFetcher，你需要在 initTask 这个初始任务里完成注册，紧接着是事件发生后分发任务的函数 EventDispatcher，它将事件按照注册时留下的函数，将发生事件后处理事件的任务扔给 taskList，并一个一个跟着处理，taskList 这样的容器可能是按照优先级优先弹出的优先队列或者其他什么算法，用以高效的分配任务，这样的任务我们给它一个名字，叫做回调函数

回调函数何时执行是由调度器决定的，原本把它放到 taskList 的函数大概率早就已经执行下去了，并不会管回调函数什么时候运行，或者说对于程序员来说，它何时运行逻辑都必须正确，这种逻辑上他俩执行没什么关系的行为就是所谓异步。

要注意的是，输入 taskList 的是一个闭包，如果假使说这个任务是一个计时器，也许你会需要循环往复的计时，此时你必须在回调函数里将计时器重新启动，启动的过程就是向 EventFetcher 再注册一遍计时器，向 EventDispatcher 再把自己的任务和自己当前正在执行的这个回调组成的一对数据再次传入，以此实现无限计时的效果

回调式的写法在现代的 UI 框架里相当常见，例如 Qt 的信号槽机制，本质上就是这样的东西，我们来看看下面 Dart 的事件循环实现

哎哎，这不就是一模一样，所以说自 IO 多路复用出现后，事件循环也很自然的就出现了

然而大量使用回调引发了回调地狱的问题。异步最大的问题就是难以阅读。异步的行为导致无法定位事件的来源，所有的任务的调用栈将都被污染为事件循环所执行的，形成了可悲的厚障壁—— asynchronous gap，为程序员调试带来了极大的困难。为了缓解回调地狱，Javascript 带头引入了 promise 的概念，但仍然未脱离回调的意图，只是让回调写成链式调用更好看一点罢了，在此处我们按下不表，有兴趣的可以自行阅读 Promise - JavaScript | MDN

横空出世的协程，async/await 的秘法

想出协程，async/await 这套东西的人绝对是个天才，我无缘得见是什么样的心路历程能够想到这一套东西，但我们仍然可以看看它是怎么工作的，这里我们主要看看的是 async/await，协程的概念与其基本一致，想必聪明的读者看完下面这些，也就能理解协程了

当我们第一次接触 async/await 时，一个最直观的困惑就是：await 字面意思就是”等待”，等待一个任务完成，这听起来不就是阻塞吗？如果它阻塞了当前线程，那和我们直接用同步的 thread.sleep() 或等待一个锁有什么区别？这岂不是换汤不换药？

实则不然。这正是 async/await 的巧妙之处，也是理解其非阻塞本质的关键。await 的”等待”，是一种主动的让步，而非被动的阻塞。

await 的”等待”是假象，状态机的”挂起”才是真相

要理解这一点，我们必须揭开 async 函数的华丽外衣，直视其内核：一个由编译器自动生成的状态机。

一个普通的同步函数从开始到结束，其执行路径是唯一的、连续的。而一个 async 函数内部因为有了 await 点，它的执行可以被分割成多个阶段。每一个 await 都是一个潜在的中断点。

让我们用一个例子来将这个状态机可视化：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
async fn fetch_data_and_process() -> String {
    let url1 = "http://example.com/data1";
    let url2 = "http://example.com/data2";

    // 阶段 0: 开始
    let data1 = fetch_url(url1).await?; // <- 等待点 1 (await 1)
    // 阶段 1: 收到 data1 后继续
    println!("Received data1: {}", data1);

    let data2 = fetch_url(url2).await?; // <- 等待点 2 (await 2)
    // 阶段 2: 收到 data2 后继续
    println!("Received data2: {}", data2);

    // 阶段 3: 结束
    format!("{} + {}", data1, data2)
}

编译器会将这个 async fn 编译成一个实现了 Future trait 的状态机。这个状态机大致长这样：

  • 基本概念 首先编译器会把 async 函数编译成状态机，在 await 发生的时候，状态机整个休眠，在外部，通过 poll 唤醒状态机，poll 的语义是尽可能催促状态机执行的更远
  • 状态 0 (State 0): 刚被调用 poll 试图推动状态机时，启动 fetch_url(url1) 这个 Future，然后立即返回 Poll::Pending。同时，它记录下自己的状态：”我现在正卡在 await 1 这里”。
  • 事件与唤醒： 当 fetch_url(url1) 的 I/O 操作在后台完成时（例如，通过操作系统的 epoll/kqueue 等机制通知），一个事件被触发。调度器会知道：”哦，fetch_url(url1) 这个 Future 准备好了”。然后，调度器会找到正在等待它的 fetch_data_and_process 状态机，并再次调用它的 poll 方法。
  • 状态 1 (State 1): 状态机从上次中断的地方恢复。它发现自己是处于”状态 1”，于是从 fetch_url(url1) 中拿到结果 data1，执行 println!，然后启动 fetch_url(url2)，接着再次返回 Poll::Pending，并将状态更新为”卡在 await 2“。
  • 状态 2 (State 2): 同理，当 fetch_url(url2) 完成时，调度器再次唤醒它。状态机恢复，拿到 data2，执行最后的逻辑，最终返回 Poll::Ready(result)，标志着整个异步任务的完成。

哦？调度器，事件的概念又被提及了，是不是事件循环是 async/await 的重要实现方式之一呢？

你猜的没有错，事实上，你可以从事件循环上很容易的构建出一个 async/await 系统来：如果所有的 async 状态机都是一个对象，当某个 async 函数运行到 await 状态的时候，就是把一个任务注册到事件循环，这个任务是，监听到” await 等待的状态机运行完毕”这个事件时，事件循环的调度器回调 poll 函数来推动自己进行下一步。而正是因为回调函数都会坍缩为 poll ，这就使得在一定程度上事件循环的任务列表里基本都是相同的 poll 函数，简化一下可以省略调任务列表，直接把发送” await 等待的状态机运行完毕”事件改成 poll ，所以 async/await 实现也有可能不需要依赖事件循环。

所以，await 的关键在于：当任务需要等待（比如 I/O 未就绪）时，它不是傻傻地占着线程不放，而是通过返回 Pending 来释放线程的控制权。这个线程此时就可以掉头去执行其他已经就绪的任务（比如另一个完成的 HTTP 请求）。 这正是异步编程高并发能力的根源。

异步任务树与调度器的工作流

想象一下，对于 ui 框架，显然要 await 的东西不可能只有一个按钮别的不等了，所以，async 需要能等待多个任务，因此你会在 rust 的宏里看见有实现这个功能的 await (join_all(v).await)，只有使用这种特制的 await，实际上是让事件循环用不同的监听模式来 poll，才能实现让一组任务异步进行。单个 async 函数是一个状态机，而一个复杂的异步程序则是由无数个 async 函数通过 await 组合而成的一棵异步任务树。

注意：异步任务树实为本文自创的概念，实际上怎么称呼这样的结构我并不知道。同样的，可能存在更复杂的数据结构能更高效的解决这样的问题，但下面我所提及的应该是一种实现。

其所有叶子节点都是 await 函数的尽头，即具体等待一个任务，树的下方是调度器，当事件发生时，调度器将会在事件列表中挑选一个事件，poll 这个事件对应的叶子节点，是一个 async 函数/协程，让它执行至挂起或结束，控制权回到调度器，处理下一个事件，如果事件消费完毕，则继续挂起，这样的循环构成了异步模式下，无栈协程组成程序的图景。

我们可以从刚刚那个 fetch_data_and_process() 为例子来画一下这棵树具体可以如何工作的其中一种样子：

首先是 fetch_data_and_process() 被 await 之前，先是有一棵树

然后紧接着，调用者开始 await，实际上干了4件事情

  1. 初始化 fetch_data_and_process 的状态机，同时包括分配返回值所放置区域等行为
  2. 把推动 fetch_data_and_process 的状态机的 poll 函数放进事件循环的任务队列
  3. 把监听 fetch_data_and_process 的状态机完成的事件注册到事件循环，回调是 poll 调用者自己
  4. 调用者挂起自己

于是便得到了

记录一下此时此刻事件循环的状态(省略无关项，同时列表的内容是伪代码，下同)：

  • 任务列表：fetch_data_and_process.poll()
  • 注册的事件：
    Event<fetch_data_and_process.finish()> -> caller.poll()

接着等待事件循环调度，直到调用 poll 推动fetch_data_and_process 的状态机，运行下来就到了 let data1 = fetch_url(url1).await?; 这一行了，也是同理，做这四件事情

事件循环的状态：

  • 任务列表：fetch_url.poll()
  • 注册的事件：
    Event<fetch_data_and_process.finish()> -> caller.poll(),
    Event<fetch_url.finish()> -> fetch_data_and_process.poll()

我们假定 fetch_url 要直接做一个获取资源的基础 API，那么它和上述的 await 实际上理应是有差别的，因为它不能再把任务推给别人了自己只是等待了，必须自己处理，所以这个异步函数的函数体里只会干两件事

  1. 创建获取 url 资源的事件，注册到事件循环，回调是 poll fetch_url 自己
  2. 挂起自己，等待数据完成

此时此刻事件循环的状态：

  • 任务列表：空
  • 注册的事件：
    Event<fetch_data_and_process.finish()> -> caller.poll(),
    Event<fetch_url.finish()> -> fetch_data_and_process.poll(),
    Event<API_getUrl.finish()> -> fetch_url.poll()

好了，接下来任务简单了，等待回调，Event<API_getUrl.finish()> 发生时，fetch_url 会被推动到结束，在结尾的时候，也会挂起自己这个状态机，运行过程回到调度器里，从而从这棵树上脱落。在此之前释放一个 Event<fetch_url.finish()> 的事件信息，告诉事件循环 fetch_data_and_process.poll() 该被放进队列里运行了。

然后 fetch_data_and_process.poll() 让 async 函数 fetch_data_and_process 继续运行，碰到第二个 await let data2 = fetch_url(url2).await?;，逻辑是一样的。直到最后 fetch_data_and_process 也执行到最后，挂起自己交给调度器，脱落，也便回到 caller 了。

这么看来实现这么一套异步的库也不是很麻烦嘛，虽然为 IO 事件包裹一层 API 是实打实的 dirty work

通过这个小例子你也可以自然的发现，如果你的应用只是普通的 await function 组合，没有特制的等待好几个任务的 await，那你的异步任务树就是异步任务链表，在这种情况下，实际上和同步毫无区别，而如果你可以同时 await 好几个任务，那由于任务是递交给事件循环穿插运行的，也便实现了所谓的异步：最简单的例子，你可以同时异步高效的 fetch 好几个 url 的内容了。

无法逾越的鸿沟：异步的”传染性”

理解了上述机制，我们也就能明白为什么说 async 函数具有”传染性”：一个函数内部如果使用了 await，它自身就必须被标记为 async。

这是因为，await 的语义依赖于其所在的上下文是一个可以被挂起和恢复的状态机。一个普通的同步函数不具备这种能力。它一旦开始执行，就必须运行到结束，中间无法被”挂起”并将线程控制权交还给调度器。

因此，异步模式是构建在同步模式之上的一种特殊环境。当你选择踏入这个环境（调用一个 async 函数），你就必须遵守它的规则。这种”鸿沟”是异步编程模型本身的性质所决定的，而非 Rust 或其他语言的设计缺陷。

本节我们聚焦于单线程内的协程调度，揭示了 async/await 非阻塞的秘密。然而，协程的魅力远不止于此。它并不与异步强绑定，其本质只是一个可暂停、可恢复的代码块，是”程序是状态机”这一最本质概念的体现。在多线程的舞台上，协程还能演绎出更精彩的并发模式。

后日谈

说真的，我觉得异步真的不是一个很好理解的东西，它和操作系统相关性不小，但你不能把它看作是操作系统的调度模式。因为操作系统作为一个具有最高权限的软件，它作为状态机，推动它的是硬件中断，在计算机模型里，它不需要考虑自己去实现这个”异步”的功能。但在实际的应用里，其实完全是同步模式，靠获取 os 的信息手动实现的推动状态机，才真正进入了异步。值得一提，我的朋友说：”其实直到在单片机里应用这些行为才真正理解了什么是异步。” 我想这正是轮询，非阻塞 IO 启发了他吧。
本人愚钝，这个概念确是想了不少时间才得以理解，中间甚至有过很久的错误理解，如有谬误，还请多多包涵