最近RTOS综述难产，估计还要花上好久才能完成初稿，就先写一篇小文练练笔。今天就给大家介绍一种常见的异步编程的范式，完成队列（Completion Queue）模型。

这个模型可能不会出现在一些设计模式的书上，我第一次见到这种模式是在研究grpc c 的异步接口时。第一次用这个模式编程感觉非常的别扭，你要手工的维护各种状态（Idle / Processing / Responding / Error / ...）而通常来说在同步模式里，这些状态的维护都算是在代码语法层面的，比如try - catch结构等。当时“年幼无知”的我还在github上写个issus吐槽grpc团队在搞啥，能不能正常写代码了。

不过随着不断深入的使用这个API，我意识到这个模式的威力是巨大的，总结就是：简单-强大-高效。大家可能好奇到底是啥，能同时满足这几个条件，那么下面我就详细地讲讲这个东西的来龙去脉。而且随着我对这个模式的理解，我在想可能很多优秀的并发框架在底层就是类似这样的机制实现的，C 的API是非常primitive的API所以会暴露这种底层实现逻辑，虽然和平常编程习惯有差异，但是能学习一下也很棒。

同步并发模式

在讲解完成队列模式之前我们先看看最最最最直接的并发模式，这个模式应该是大家在学校里学习多线程的时候都会学习到的一种写法（尤其是学习操作系统原理时）。

伪代码大概如下：

func process(request, response) { try { ...处理request... } catch { ...写入错误到response... } finally { ...写入结果到response... }} func threadWorker(queue) { while (!exiting) { request, response = queue.wait_pop() process(request, response) }}func mainLoop(listener, queue) { while (!exiting) { request, response = wait_for_request(listener) queue.push(request, response) }}func main() { listener = newListener() queue = newQueue() threads = [] for (i = 0; i < cpu_num; i ) { threads.push(Thread(threadWorker, queue)) } mainLoop(listener, queue)}

这个模式非常简单清晰，每个线程运行一个函数等待新的请求，队列里有新的请求进来就调用处理函数之后等待下个请求。主线程等待新的请求，有请求之后就写到一个队列里交给各个线程执行。

这个是非常经典的Producer - Consumer模式，是练习各种数据结构、学习操作系统线程、同步互斥手段特别好的例程。

但是！这个模式的性能（在绝大部分情况下）是非常糟糕的！我们来看看到底是哪里出了问题。

首先我们假设请求量是无穷无尽的，每时每刻都有很多的请求进来，那么影响这个服务处理能力的瓶颈就是这个服务能同时处理多少任务，以cpu及每个任务的处理速度。而在业务逻辑、算法实现不变的情况下，一个次请求需要的「累积CPU时利用率间」是基本一致的。注意：这里用了「累积CPU时间」这个名词，需要特别说明一下，一段完整的业务逻辑在执行时会分成多次执行，造成这种现象的原因包括抢占、系统调用、io等等。所以这里计算的是实际有效利用CPU计算的时间而不是物理时间。

那么在这个前提下，影响服务性能的因素就是可以同时处理的业务数量。有的同学可能会问：CPU的一个核心同时只能执行一段代码，以上代码已经生成了「CPU核心」个线程同时执行，肯定是完整利用了CPU的计算资源了。

这个说法也对也不对，在绝大部分情况下是不对的 —— 只有在一段业务逻辑没有任何系统调用完全是计算逻辑时才能占满全部的CPU计算资源，一旦涉及到系统调用、io就会出现CPU闲置的情况。

那么有的同学又会问：我的代码写得特别好，纯计算逻辑，这样就没问题了吧？答案还是：No！我们把实际场景代入进来：首先服务要从网络中读取完整的请求，然后计算结果，再把结果完整地写到网络设备 —— 显而易见这个过程至少有两次io操作：从网络设备读取和写入网络设备。在以上的实现方式中，读取和写入这两个过程仍然会造成CPU闲置。当然这个场景也有一些在不改变同步模式的情况下的优化手段，比如全部用缓冲区来代替真实的读取、写入过程。不过一般来说缓冲区会作为一个对「io低速设备的性能优化」而非针对「糟糕的设计模式的优化」，不太可能完成解决这个问题（比如一个请求payload 500MB的请求……不太可能完全写入缓冲区）。

那么了解以上的知识之后，大家应该基本认可了即使是最简单的业务逻辑，同步模式仍然会有巨大的性能损耗，CPU不能保证时时刻刻处在100%利用率，浪费了计算资源。（什么，怕CPU过热？确实有这个问题，所以各个大厂在给机器降温这个问题上也有很多花招，不过这里就不说了，另外实际工业环境中确实不会让CPU一直处在100%运行）

现实的业务逻辑往往更加复杂，需要调用很多后端接口（一次调用至少2次io），系统调用（中断内执行复杂任务）等等，因此性能会随着业务逻辑的复杂越发的劣化。

到这个时候，有一些聪明的同学会提出：我有办法了！多申请一些线程就行了！很多技术文档不是说申请cpu_num * 2的线程最佳吗？！只要一个线程进入等待CPU闲置阶段，操作系统就会寻找下一个可以执行的线程，利用率不就高了吗？

嗯，能想到这一点说明你的基础很扎实，事实上没错，这个方法确实能提升性能。但是申请更多线程是有代价的，这些代价包括：

• 在内核维护更多线程数据，一些框架实现了用户态线程，那么也会相应地增加内核、用户态的线程结构大小
• 内核线程切换的成本是不小的，需要切换内存地址的标志符，切换执行栈，保存、弹出寄存器值等等
• 现代CPU往往有很大的L1/L2/L3缓存（没错，这些缓存的速度比内存快得多的多的多的多），一段程序运行时如果不切换运行核心那么数据命中缓存的概率是很大的，如果频繁切换核心是比较糟糕的。当然你可以把线程绑定在一个核心上，不过这个就会造成各个核心负载不均了

那么这个问题可以解决吗？当然可以了！完成队列出场了！

完成队列怎么模式

完成队列模式和我们常说的回调模式在实现效果上几乎是一致的，回调模式是可以以完成队列模式为基础严格地实现出来的。另外一些语言提供了async / await等等机制，有的框架实现方式也是把代码切分成了几个不同的回调来执行，即使在你看起来这些代码是写在一个函数里的。

那么说了这么多，完成队列的代码到底长啥样？下面我也是写一段伪代码（完成队列的核心是状态，写法有很多种）：

// 一个Processor对象只处理一次请求class Processor { *cq, // 完成队列（completion queue） state = idle, // 当前执行状态 request, response, context, // 执行过程中的参数保存 func process() { if (state == Initial) { // 初始化状态，等待请求 process_idle_state() } else if (state == New) { // 收到新的请求状态 process_new_state() } else if (state == Prepared) { // 新请求接收完毕 process_prepared_state() } else if (state == SomeState1) { // 内部状态1 process_some_state1() } else if (state == SomeState2) { // 内部状态2 process_some_state2() } else if (state == Finalize) { // 请求完成 process_Finalize_state() } } func process_idle_state() { // 接收新的请求 // 注意：这个函数不会导致当前函数进入阻塞状态，执行完毕后直接返回，此时新的请求可能还没来。 receive_request(&request, &response, cq, this) // 设置新的状态是接收到新请求 state = New } func process_new_state() { // 接受到了新的请求 // 创建一个新的Processor，接收下一个请求 (new Processor(cq...)).process() // 接收请求数据 receive_request_payload(&context[request_payload], cq, this); // 设置新的状态是接收完毕 state = Prepared } func process_prepared_state() { // 新请求接收完毕，执行业务逻辑 ...request, context[request_payload]... // 调用API 1 // 注意：执行到这里之后函数就返回了，并不会等待API返回 call_api_1(&context[api1_request], &context[api1_response], cq, this) // 设置新的状态是接收新请求 state = SomeState1 } func process_some_state1() { // API1执行完毕 ...context[api1_response]... // 调用API 2 // 注意：执行到这里之后函数就返回了，并不会等待API返回 call_api_2(&context[api2_request], &context[api2_response], cq, this) // 设置新的状态是接收新请求 state = SomeState2 } func process_some_state2() { // API2执行完毕 ...context[api2_response]... // 业务逻辑完成，写入response ...response... write_response(..., cq, this) // 设置新的状态是完成 state = Finalize } func process_finalize_state() { // 写入response完成，清理本次请求使用的所有资源 close(...) delete this }}func threadWorker(cq) { // 创建一个完成队列 cq = new CompletionQueue() // 创建一个执行器，接收请求 (new Processor(cq...)).process() while (!exiting) { // // 这个地方大家可能看不太明白，我解释一下，在上文中调用的方法： // receive_request // call_api_1 // call_api_2 // write_response // 这四个方法都有一个特点，就是后两个参数是this (Processor对象自身）和cq（CompletionQueue对象） // 这四个方法在执行完毕后，会执行cq.push(processor)，就是这里的代码cq.next()返回的processor， // 然后调用process()方法 // // 这里大家可能还有一个问题，这四个方法执行的结果到哪里去了？可以大家再回头去看这些函数调用过程实际上 // 是传入了response对象的，函数执行的结果是写入到这个对象里的。这也是context字段存在的主要目的之一， // 就是保存中间执行结果。 // processor = cq.next() processor.process() }}func main() { threads = [] for (i = 0; i < cpu_num; i ) { // 启动线程 threads.push(Thread(threadWorker, queue, cq)) } start_server()}

以上逻辑看起来比较长，主要是解释了各个阶段的细节。完成队列没有一个标准的实现，上面只是最简单的一种，我大概讲一下流程：

• 首先创建cpu_num个线程，每个线程创建一个独立的CompletionQueue，并且启动第一个Processor接收请求。注意此时CPU是闲置的。
• start_server()会启动内部逻辑，接收请求。注意，此时如果receive_request函数没有被调用，则不会有请求被处理。
• 当receive_request被调用后，server内部逻辑会将接收到的请求发给对应的cq
• threadWorker不断的从cq中获得下一个准备好的任务执行

这个设计与同步模式最大的差异就是，但凡涉及到长时间的io操作，执行流程都要中断到这一时刻，把CPU时间让出，待io操作完成之后通过写入completion queue的方式继续执行下一个状态的逻辑。

因此可以看到只要调用后端的API过程没有阻塞CPU浪费时间的实现，那么整个流程几乎不会浪费计算资源，CPU一直处在满负荷计算的状态。

那么这个模式是否有什么弊端呢？答案一定是有的，一些同步模式遇到的问题完成队列依然会遇到：

• 一个线程里的执行器发生阻塞，会导致各个线程负载不均
• CPU切换线程执行的核心导致CPU缓存失效

这两个问题在工业实现层面是被优化了的，第一个问题是可以靠动态地切换任务所在的队列可以缓解（不过这个方法是比较危险的，因为他会导致所有tls「Thread local storage」失效和错误）。CPU切换执行核心的问题绑定核心即可。

在实际开发过程中，一般来说大家不太需要真的从0开始实现CompletionQueue，有大量的框架在这个之上提供了更加便捷的API，有一些语言甚至提供了语法层面的抽象封装。

比如我请大家思考一个问题，如果以上的例子中每个process_xxx都是一个回调函数，是不是有点儿像Promise Fetch的API了？ ：）

当然如果你要写一个底层框架，那么完成队列是一个非常不错的起点。

以上就是本次对完成队列的全部介绍了，这个只是一个非常简单的介绍，要实现一个好的CompletionQueue需要非常多的思考和细节上的打磨，大家如果感兴趣可以自己写着试试了！