徹底理解協程
1 詳解協程
1.1 多線程的困境
人類壓榨CPU的腳步從未停止過。在實際的生產過程中，我們將CPU的任務分為兩大類：
- 計算密集型：數值計算、邏輯判斷的任務較多。CPU利用率非常高。
- IO密集型：與IO設備交互，如讀取磁盤和網卡，頻繁等待IO操作結果。CPU利用率非常低。
為了提高IO密集型任務的CPU利用率，常常采用異步加回調的方案。我們去餐廳吃飯，點菜之后就可以回座位上刷手機了，這叫異步；飯菜做好了，服務員把菜端過來，這叫回調。
在軟件開發的過程中，異步加回調的方案將一件事拆成兩個過程，不符合人類的線性思維，增加了代碼復雜度，提高了排查錯誤的難度。這就好比，我們下單后回座位等待，雖然有空干別的事情，但是也不能離開餐廳，心里要記得菜還沒上。
最簡單的方法是，下單之后在窗口等著，直到廚師做好了，我們才端走飯菜，這叫做同步阻塞。同步阻塞的方案簡單直接，程序員的心智負擔最輕，如下代碼所示：
/**
* 顧客用餐
*
* @param customerOrder 顧客訂單
* @return
*/
public void customerDish(CustomerOrder customerOrder) {
// 顧客下單，生成訂單
RestaurantOrder restaurantOrder = submitOrder(customerOrder);
// 廚房接到訂單，開始做飯，耗時5分鐘
CustomerDish customerDish = cookCustomerDish(restaurantOrder);
// 顧客拿到飯菜，開始吃飯
customerEating(customerDish);
}
如果很多顧客來吃飯，都聚集在窗口等待，相當于將處理過程變為線程，放入線程池中執行，如下代碼所示：
/**
* 顧客吃飯的線程
*/
class CustomerDishThread extends Thread {
private CustomerOrder customerOrder;
CustomerDishThread(CustomerOrder customerOrder) {
this.customerOrder = customerOrder;
}
@Override
public void run() {
// 顧客用餐
customerDish(customerOrder);
}
}
private static final ExecutorService THREAD_POOL = Executors.newCachedThreadPool();
/**
* 餐廳接待很多顧客
* @param customerOrderList
*/
public void serveManyCustomer(List<CustomerOrder> customerOrderList) {
for (CustomerOrder customerOrder : customerOrderList) {
THREAD_POOL.execute(new CustomerDishThread(customerOrder));
}
}
同步阻塞方案是低效的，浪費顧客的時間，窗口也擠不下太多人。如果把餐廳看作服務端，把顧客看成客戶端的請求，服務端能夠并發執行的線程數有限。當線程非常多的時候，操作系統頻繁調度線程，上下文切換是不小的開銷。有沒有辦法減少線程調度的開銷呢？
協程登場了。
1.2 協程的優勢
協程（Coroutines）的完整定義是“協作式調度的用戶態線程”。首先，要理解線程調度的兩種方式：
-
協作式調度：當前線程完全占用CPU時間，除非自己讓出時間片，直到運行結束，系統才執行下一個線程。可能出現一個線程一直占有CPU，而其他線程等待。
-
搶占式調度：操作系統決定下一個占用CPU時間的是哪一個線程，定期的中斷當前正在執行的線程，任何一個線程都不能獨占。不會因為一個線程而影響整個進程的執行。
另外，要理解用戶態和內核態的概念。
操作系統的核心是內核(kernel)，它獨立于普通的應用程序，可以訪問受保護的內存空間，也有訪問底層硬件設備的所有權限。有些CPU 的指令是非常危險的，一旦用錯可能導致系統崩潰。如果所有的程序都可以任意使用這些指令，那么系統崩潰的概率將大大增加。為了保證內核的安全，操作系統一般都禁止用戶進程直接操作內核。具體的實現方式是將虛擬內存空間劃分為兩部分，一部分為內核空間，另一部分為用戶空間。當進程運行在內核空間時就處于內核態，進程運行在用戶空間時則處于用戶態。
無論是進程還是線程，它們的上下文切換和"內核態、用戶態"沒有直接的關系。比如只要需要系統調用，即使不做任何切換，都需要進入內核態。舉個例子：一個線程調用函數在屏幕上打印 hello world，就已經進入了內核態了，因為打印字符的功能是由內核程序提供的。總的來說，應用程序通常運行在用戶態，遇到下列三種情況會切換到內核態：
- 系統調用：創建和調度線程、加鎖解鎖等等。
- 異常事件：發生不可知的異常時切換到內核態，以執行相關的異常事件。
- 設備中斷：如果外圍設備完成了用戶請求，比如硬盤讀寫操作，就會給CPU發送中斷信號。CPU會轉去處理中斷事件，切換到內核態。
線程的代碼在用戶態運行，而調度是在內核態運行的。操作系統切換線程上下文的步驟如下所示：
- 1）保留用戶態現場（上下文、寄存器、用戶棧等）
- 2）復制用戶態參數，用戶棧切到內核棧，進入內核態
- 3）代碼安全檢查（內核不信任用戶態代碼）
- 4）執行內核態代碼
- 5）復制內核態代碼執行結果，回到用戶態
- 6）恢復用戶態現場（上下文、寄存器、用戶棧等）
協程不是操作系統的底層特性，系統感知不到它的存在。它運行在線程里面，通過分時復用線程的方式運行，不會增加線程的數量。協程也有上下文切換，但是不會切換到內核態去，比線程切換的開銷要小很多。每個協程的體積比線程要小得多，一個線程可以容納數量相當可觀的協程。
在IO密集型的任務中有著大量的阻塞等待過程，協程采用協作式調度，在IO阻塞的時候讓出CPU，當IO就緒后再主動占用CPU，犧牲任務執行的公平性換取吞吐量。
事物都有兩面性，協程也存在幾個弊端：
- 線程可以在多核CPU上并行，無法將一個線程的多個協程分攤到多核上。
- 協程執行中不能有阻塞操作，否則整個線程被阻塞。
- 協程的控制權由用戶態決定，可能執行惡意的代碼。
1.3 協程的原理
無論是線程還是協程，都只是操作系統層面的抽象概念，本質是函數執行的載體。可以簡單的認為協程是一個能夠被暫停以及被恢復運行的函數，在協作調度器的控制下執行，同一個時刻只能運行一個函數。
我們來看看下面的Java代碼，代碼中出現的注解 Coroutine 和 CoroutineSchedule ，只是為了更好的演示而編造出來，JDK并沒有這兩個注解。
public class CoroutineDemo {
static void functionA() {
System.out.println("A");
}
static void functionB() {
System.out.println("B");
}
static void functionC() {
System.out.println("C");
}
/**
* 普通的函數
*/
static void commonFunction() {
functionA();
functionB();
functionC();
}
/**
* @Coroutine 標識函數為協程
*/
@Coroutine
static void coroutineFunction() {
functionA();
functionB();
functionC();
}
/**
* @Coroutine 標識協程調度器，跟隨主線程一起啟動
*/
@CoroutineSchedule
void coroutineScheduleRule() {
//如果等待IO，暫停協程coroutineFunction，否則就恢復
if(waitIO()){
yieldFunction("coroutineFunction");
}else {
resumeFunction("coroutineFunction");
}
}
public static void main(String[] args) {
Thread commonThread = new Thread(() -> {
//執行普通函數
commonFunction();
});
commonThread.start();
Thread coroutineThread = new Thread(() -> {
//執行協程
coroutineFunction();
});
coroutineThread.start();
}
}
main方法啟動了兩個線程，普通函數 commonFunction 執行后，會依次打印出 A B C。協程 coroutineFunction 執行后，不確定打印什么，因為協程調度器有規則：如果CPU繁忙就暫停協程。如果協程打印了 A 之后就被暫停了，當它被再次喚醒，可能會接著打印 B C，而不是打印 A 。因為協程記錄了函數執行的上下文信息，知道自己上一次執行到了哪里。這和操作系統調度線程是一樣的，暫停當前線程，保存運行狀態后去調度其它線程，該線程再次被分配CPU后繼續運行，就像沒有被暫停過一樣。
1.4 如何實現協程
我們嘗試一下用 C/C++ 實現一個簡單的協程。協程有兩個重要的部分：調度器和用戶態的上下文切換。Linux系統已經提供了操作用戶態上下文的接口，只需要實現調度器即可。glibc是一個C語言庫，封裝了系統最重要的系統服務，提供了最底層的API。glibc包含一個ucontext庫，支持用戶態的上下文切換。
首先看看ucontext提供的四個基本函數：
| 函數 | 作用 |
|---|---|
| int getcontext(ucontext_t *ucp) | 獲得當前上下文保存的棧和入口執行點 |
| int setcontext(const ucontext_t *ucp) | 設置當前上下文。初始化ucp結構體，將當前的上下文保存到ucp中 |
| void makecontext(ucontext_t ucp, void (func)(), int argc, ...) | 創建一個新的上下文。修改上下文ucp，給該上下文指定一個棧空間ucp->stack，設置后繼的上下文ucp->uc_link |
| int swapcontext(ucontext_t *oucp, ucontext_t *ucp) | 切換上下文。保存當前上下文到oucp，設置到ucp所指向的上下文，跳轉到ucp所指的地方 |
ucontext_t 是用戶態上下文數據，看看它的數據結構：
typedef struct ucontext {
// 后繼的上下文，表示當前程序執行之后下一個上下文
struct ucontext *uc_link;
sigset_t uc_sigmask;
// 上下文堆棧
stack_t uc_stack;
mcontext_t uc_mcontext;
} ucontext_t;
在下面的代碼演示中，你將會進一步理解這4個函數的用法，代碼的調試環境是Ubuntu 16、Visual Studio Code（包含 C/C++ 開發插件）：
#include <stdio.h>
#include <ucontext.h>
#include <unistd.h>
int i = 1 , max = 5;
int main() {
ucontext_t context;
puts("上菜了");
getcontext(&context);
if (i > max ) return 0;
puts("張三吃飯了");
i++;
setcontext(&context);
puts("李四吃飯了");
return 0;
}
李四吃上飯嗎？你大概能夠猜到代碼不會執行到puts("李四吃飯了");，以上代碼的輸出結果是：
上菜了
張三吃飯了
張三吃飯了
張三吃飯了
張三吃飯了
張三吃飯了
getcontext(&context)獲取了程序執行的上下文，setcontext(&context)給當前程序設置上下文，程序立即重新執行。&context記錄了已經執行的代碼行，那么再次執行的起始行是if (i > max ) return 0，這樣永遠不會走到puts("李四吃飯了")。
以下代碼演示了 makecontext 和 swapcontext 函數的用法，以及設置上下文堆棧參數：
#include <ucontext.h>
#include <stdio.h>
void eating()
{
puts("李四吃飯了");
}
int main()
{
//指定棧空間
char stack[512*128];
ucontext_t child,main;
//獲取當前上下文
getcontext(&child);
//指定棧空間
child.uc_stack.ss_sp = stack;
//指定棧空間大小
child.uc_stack.ss_size = sizeof(stack);
child.uc_stack.ss_flags = 0;
//設置后繼上下文
child.uc_link = &main;
puts("上菜了");
//修改 child 上下文，指向eating函數
makecontext(&child,(void (*)(void))eating,0);
//切換到child上下文，保存當前上下文到main
swapcontext(&main,&child);
puts("張三吃飯了");
return 0;
}
以上代碼的輸出結果是：
上菜了
李四吃飯了
張三吃飯了
入口main方法是一個線程，函數swapcontext(&main,&child)交換了上下文參數，將會執行函數eating()，之后再執行child的后繼上下文main，回到了主線程main。從這段代碼你能否想到如何實現一個協程調度器？
在真實的生產環境下，協程調度器是個運行在后臺的線程，自動化調度所有協程，調度規則也比較復雜。以下代碼將實現一個無法自動化調度的調度器。
首先定義協程結構體：
//上下文堆棧
#define DEFAULT_STACK_SZIE (512*128)
//定義協程狀態
enum ThreadState{FREE,RUNNABLE,RUNNING,SUSPEND};
//定義協程結構體
typedef struct uthread_t
{
ucontext_t ctx;
Fun func;
void *arg;
enum ThreadState state;
char stack[DEFAULT_STACK_SZIE];
}uthread_t;
定義調度器結構體:
//最大協程數量
#define MAX_UTHREAD_SIZE 512
typedef struct schedule_t
{
ucontext_t main;
//正在運行的協程的ID，一個線程只能運行一個協程
int running_thread;
uthread_t *threads;
//協程數量
int uthread_count;
schedule_t():running_thread(-1), uthread_count(0) {
threads = new uthread_t[MAX_UTHREAD_SIZE];
for (int i = 0; i < MAX_UTHREAD_SIZE; i++) {
threads[i].state = FREE;
}
}
~schedule_t() {
delete [] threads;
}
}schedule_t;
定義協程調度方法:
// 創建協程
int uthread_create(schedule_t &schedule,Fun func,void *arg);
// 掛起協程
void uthread_yield(schedule_t &schedule);
// 恢復協程
void uthread_resume(schedule_t &schedule,int id);
實現協程調度方法:
// 創建協程
int uthread_create(schedule_t &schedule, Fun func, void *arg)
{
int id = 0;
for (id = 0; id < schedule.uthread_count; ++id)
{
if (schedule.threads[id].state == FREE)
{
break;
}
}
if (id == schedule.uthread_count)
{
schedule.uthread_count++;
}
uthread_t *t = &(schedule.threads[id]);
t->state = RUNNABLE;
t->func = func;
t->arg = arg;
getcontext(&(t->ctx));
t->ctx.uc_stack.ss_sp = t->stack;
t->ctx.uc_stack.ss_size = DEFAULT_STACK_SZIE;
t->ctx.uc_stack.ss_flags = 0;
t->ctx.uc_link = &(schedule.main);
schedule.running_thread = id;
//創建協程結構體
makecontext(&(t->ctx), (void (*)(void))(uthread_init), 1, &schedule);
//切換上下文，執行func函數
swapcontext(&(schedule.main), &(t->ctx));
return id;
}
//初始化一個協程，配合uthread_create使用
void uthread_init(schedule_t *ps)
{
int id = ps->running_thread;
if (id != -1)
{
uthread_t *t = &(ps->threads[id]);
t->func(t->arg);
t->state = FREE;
ps->running_thread = -1;
}
}
// 恢復執行協程
void uthread_resume(schedule_t &schedule, int id)
{
if (id < 0 || id >= schedule.uthread_count)
{
return;
}
uthread_t *t = &(schedule.threads[id]);
if (t->state == SUSPEND)
{
// 上下文切到t->ctx，即恢復執行協程
swapcontext(&(schedule.main), &(t->ctx));
}
}
// 掛起協程
void uthread_yield(schedule_t &schedule)
{
if (schedule.running_thread != -1)
{
uthread_t *t = &(schedule.threads[schedule.running_thread]);
t->state = SUSPEND;
schedule.running_thread = -1;
// 上下文切回主線程，相當于掛起協程
swapcontext(&(t->ctx), &(schedule.main));
}
}
測試調度方法：
void zhangsan(void * arg)
{
puts("張三吃飯了");
//掛起協程
uthread_yield(*(schedule_t *)arg);
puts("張三吃完了");
}
void lishi(void *arg)
{
puts("李四吃飯了");
//掛起協程
uthread_yield(*(schedule_t *)arg);
puts("李四吃完了");
}
int main()
{
//初始化調度器
schedule_t schedule;
//創建協程并掛起
int zhangsan_id = uthread_create(schedule,zhangsan,&schedule);
int lisi_id = uthread_create(schedule,lishi,&schedule);
//恢復協程
uthread_resume(schedule,zhangsan_id);
uthread_resume(schedule,lisi_id);
puts("餐廳營業中");
return 0;
}
以上程序的輸出結果：
張三吃飯了
李四吃飯了
張三吃完了
李四吃完了
餐廳營業中
目前許多語言已經支持協程，比如C#、Golang、Python、Lua、Ruby、C++ 20、Erlang，也有一些 C/C++ 開源的協程庫，比如Protothreads、libco。
是不是缺了一個年老色衰的Java？
2 Java協程
目前還沒有JDK正式版本支持協程特性，如果想嘗試Java的協程，可以使用Open JDK 19的預覽特性或者 Alibaba JDK 最新版，以及第三方框架Quasar。
2.1 JDK的VirtualThread
2018年1月，OpenJDK官方提出了協程項目Project Loom。2019年，Loom的首個EA版本問世，此時Java的協程類叫做Fiber。它將使用Fiber輕量級用戶模式線程，從JVM層面對多線程技術進行徹底的改變，使輕量級線程的并發也能夠適用于高吞吐量的業務場景。2019年10月，官方將Fiber重新實現為Thread的子類VirtualThread，兼容Thread的所有操作。
2021年11月15日，OpenJDK官方宣布 JDK 19中加入虛擬線程的特性 JEP 425: Virtual Threads (Preview)。
Virtual threads are lightweight threads that dramatically reduce the effort of writing, maintaining, and observing high-throughput concurrent applications. (虛擬線程是輕量級線程，可以顯著減少編寫、維護和觀察高吞吐量并發應用程序的工作量)
該特性屬于預覽版，距離穩定版本還需要一段時間。如要在 JDK 19上嘗試該功能，則必須通過--enable-preview啟動，如下所示：
java --release 19 --enable-preview Main.java
簡單了解一下VirtualThread的相關API：
// 啟動一個簡單虛擬線程
Thread thread = Thread.ofVirtual().start(runnable);
// 采用ThreadFactory創建虛擬線程
ThreadFactory factory = Thread.ofVirtual().factory();
// 創建大量虛擬線程
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
IntStream.range(0, 10_000).forEach(i -> {
executor.submit(() -> {
Thread.sleep(Duration.ofSeconds(1));
return i;
});
});
}
想了解更多細節可以閱讀 https://openjdk.org/jeps/425
2.2 Quasar框架
Quasar是一個開源的Java協程框架，基本原理是修改字節碼，使方法掛起后可以保存和恢復JVM棧幀，方法內部已執行到的字節碼位置也通過增加狀態機的方式記錄，在下次恢復后可直接跳轉到中斷的位置。項目地址是 http://docs.paralleluniverse.co/quasar/ 。
我們測試一下使用線程和協程并發執行10000次的消耗，代碼如下所示：
// 使用JDK的線程和線程池
public static void main(String[] args) throws Exception {
CountDownLatch countDownLatch=new CountDownLatch(10_000);
long start = System.currentTimeMillis();
ExecutoarService executor= Executors.newCachedThreadPool();
for (int i = 0; i < 10_000; i++) {
executor.submit(() -> {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
countDownLatch.countDown();
});
}
countDownLatch.await();
long end = System.currentTimeMillis();
System.out.println("Thread use:"+(end-start)+" ms");
}
接下來使用Quasar框架的協程，maven依賴配置：
<dependency>
<groupId>co.paralleluniverse</groupId>
<artifactId>quasar-core</artifactId>
<version>0.7.10</version>
</dependency>
JVM啟動參數要配置--javaagent:C:\Users\Administrator\.m2\repository\co\paralleluniverse\quasar-core\0.7.10\quasar-core-0.7.10.jar。
public static void main(String[] args) throws Exception {
CountDownLatch countDownLatch=new CountDownLatch(10_000);
long start = System.currentTimeMillis();
for (int i = 0; i < 10_000; i++) {
new Fiber<>(new SuspendableRunnable(){
@Override
public Integer run() throws SuspendExecution, InterruptedException {
Fiber.sleep(1000);
countDownLatch.countDown();
}
}).start();
}
countDownLatch.await();
long end = System.currentTimeMillis();
System.out.println("Fiber use:"+(end-start)+" ms");
}
以上代碼執行結果可以看出協程性能高出一倍，其他方面的比對如內存消耗、GC等，請讀者自行研究。
2.3 阿里巴巴JVM
阿里巴巴JVM團隊根據自身業務需要，在 Open JDK 的基礎上開發了Alibaba Dragonwell，該版本攜帶的Wisp2組件讓JVM支持了協程。阿里巴巴的核心電商應用已經在協程模型上經過兩個雙十一的考驗，性能和穩定性得到了驗證。
Wisp協程完全兼容現有多線程的代碼寫法，僅增加JVM參數來開啟協程。我們來嘗試一下，先通過地址 https://github.com/alibaba/dragonwell8/releases/tag/dragonwell-standard-8.12.13_jdk8u345-ga 下載dragonwell8，這個版本相當于Oracle JDK 1.8。在JVM啟動參數中增加-XX:+UseWisp2，即開啟了協程。
以下代碼演示了在線程中將2個阻塞隊列的數據交換100000次。
public class Wisp2Demo {
private static final ExecutorService THREAD_POOL = Executors.newCachedThreadPool();
public static void main(String[] args) throws Exception {
BlockingQueue<Byte> q1 = new LinkedBlockingQueue<>(), q2 = new LinkedBlockingQueue<>();
THREAD_POOL.submit(() -> loop(q2, q1));
Future<?> f = THREAD_POOL.submit(() -> loop(q1, q2));
q1.put((byte) 1);
System.out.println(f.get() + " ms");
}
private static long loop(BlockingQueue<Byte> in, BlockingQueue<Byte> out) throws Exception {
long start = System.currentTimeMillis();
for (int i = 0; i < 1_000_000; i++) out.put(in.take());
return System.currentTimeMillis() - start;
}
}
正常啟動JVM：
java Wisp2Demo
6778 ms
帶參數啟動JVM：
// UnlockExperimentalVMOptions 允許使用實驗性參數，保證UseWisp2生效
// ActiveProcessorCount 指定JVM可用的CPU數
java -XX:+UnlockExperimentalVMOptions -XX:+UseWisp2 -XX:ActiveProcessorCount=1 Wisp2Demo
690 ms
啟用協程之后觀察耗時情況，性能提升了近10倍。
參考文檔
公 眾 號：編碼專家
獨立博客：codingbrick.com
文章出處：http://www.rzrgm.cn/xiaoyangjia/p/16640741.html
本文版權歸作者所有，任何人或團體、機構全部轉載或者部分轉載、摘錄，請在文章明顯位置注明作者和原文鏈接。
浙公網安備 33010602011771號