目錄
探究 iOS 協(xié)程 - 協(xié)程介紹與使用（一）
探究 iOS 協(xié)程 - coobjc 源碼分析（二）
一.什么是協(xié)程
介紹
首先引用維基百科的一段介紹
協(xié)程是計算機(jī)程序的一類組件蒸其，推廣了協(xié)作式多任務(wù)的子程序，允許執(zhí)行被掛起與被恢復(fù)。相對子例程而言腻豌，協(xié)程更為一般和靈活座咆，但在實踐中使用沒有子例程那樣廣泛峰弹。協(xié)程源自Simula和Modula-2語言理卑，但也有其他語言支持距辆。協(xié)程更適合于用來實現(xiàn)彼此熟悉的程序組件余佃，如協(xié)作式多任務(wù)、異常處理跨算、事件循環(huán)爆土、迭代器、無限列表和管道诸蚕。
根據(jù)高德納的說法, 馬爾文·康威于1958年發(fā)明了術(shù)語coroutine并用于構(gòu)建匯編程序步势。[1] [2] 協(xié)程最初在1963年被提出。[2]
上面提到的「子例程」我們可以簡單理解為函數(shù)調(diào)用背犯。維基百科還列出了函數(shù)調(diào)用和協(xié)程的區(qū)別坏瘩，這里我總結(jié)了一下，具體如下：
- 函數(shù)調(diào)用的入口只有函數(shù)的起始位置漠魏，一旦退出就完成了函數(shù)調(diào)用倔矾。每個函數(shù)調(diào)用只會返回一次。
- 函數(shù)調(diào)用在所有語言中都是層級調(diào)用（函數(shù)調(diào)用棧）柱锹。
- 協(xié)程可以通過
yield中斷執(zhí)行哪自，轉(zhuǎn)而執(zhí)行別的協(xié)程。在這種轉(zhuǎn)換過程中不存在調(diào)用者與被調(diào)用者的關(guān)系禁熏。
簡單來說壤巷，協(xié)程（coroutine）是一種程序運行的方式，可以理解成 協(xié)作的線程 或 協(xié)作的函數(shù) 瞧毙。協(xié)程既可以用單線程實現(xiàn)胧华，也可以用多線程實現(xiàn)寄症。前者是一種特殊的子例程，后者是一種特殊的線程矩动。
歷史
徐宥博士在他的博客里有提到：
雖然協(xié)程是伴隨著高級語言誕生的瘸爽，它卻沒有能像子過程一樣成為通用編程語言的基本元素。
從 1963 年首次提出到上個世紀(jì)九十年代铅忿，我們在 ALOGL, Pascal, C, FORTRAN 等主流的命令式編程語言中都沒有看到原生的協(xié)程支持。協(xié)程只稀疏地出現(xiàn)在 Simula灵汪，Modular-2 (Pascal 升級版) 和 Smalltalk 等相對小眾的語言中檀训。協(xié)程作為一個比子進(jìn)程更加通用的概念，在實際編程卻沒有取代子進(jìn)程享言，這一點不得不說是出乎意外的峻凫。如果我們結(jié)合當(dāng)時的程序設(shè)計思想看，這一點又是意料之中的：協(xié)程是不符合那個時代所崇尚的“自頂向下”的程序設(shè)計思想的览露，自然也就不會成為當(dāng)時主流的命令式編程語言 (imperative programming) 的一部分荧琼。
但是因為硬件性能的提升、多線程等開始普及差牛，協(xié)程又重回歷史舞臺大放異彩命锄。
說一說我理解的協(xié)程
通俗一點講，協(xié)程就是可以掛起（暫停）任意函數(shù)的執(zhí)行偏化，這個掛起操作是由開發(fā)者來完成的脐恩，并且你可以在你想恢復(fù)的時候恢復(fù)它。它跟線程最大的區(qū)別在于線程一旦開始執(zhí)行侦讨，從任務(wù)的角度來看驶冒，就不會被暫停，直到任務(wù)結(jié)束這個過程都是連續(xù)的韵卤，線程之間是搶占式的調(diào)度骗污，因此也不存在協(xié)作問題。這只是協(xié)程提供的最基本的能力沈条，基于這個能力我們可以做很多事情需忿。看起來其實概念很簡單蜡歹，但是應(yīng)用起來還是比較復(fù)雜的贴谎。
二.大前端協(xié)程的發(fā)展
前端
早在2015年，JavaScript 推出 ECMAScript 6 標(biāo)準(zhǔn)的時候季稳，就引入了協(xié)程的編程方式擅这，我們可以看一個例子：
對于傳統(tǒng)的網(wǎng)絡(luò)請求，應(yīng)該是下面這個樣子
ajax({
method: 'POST',
url: url,
success: function (data) {
//解析 json
data.json(function(json) {
//在這里對獲取的數(shù)據(jù)進(jìn)行操作.....
})
})
這種異步編程方式被稱之為「回調(diào)地獄」景鼠。
為了解決「回調(diào)地獄」的問題仲翎，ECMAScript 6 引入了幾種異步編程方式痹扇。
Promise
Promise 從字面上來講，就代表一種承諾溯香。從語法上來講鲫构，它是一個對象。簡單來說玫坛，Promise 就是一個容器结笨，里面保存著某個未來才會結(jié)束的事件。Promise 內(nèi)部有三種狀態(tài)：pending(等待)湿镀，fulfiled(成功)炕吸，rejected(失敗)。Promise 的狀態(tài)只會受異步操作的影響勉痴，并且一旦狀態(tài)改變之后赫模，就不會在變。Promise 對象的狀態(tài)改變蒸矛，只有兩種可能：從 pending 變?yōu)?fulfilled 和從 pending 變?yōu)?rejected瀑罗。
來看一看 Promise 的構(gòu)造方法：
const promise = new Promise(function(resolve, reject) {
// ... some code
if (/* 異步操作成功 */){
resolve(value);
} else {
reject(error);
}
});
Promise 構(gòu)造函數(shù)接受一個函數(shù)作為參數(shù)，該函數(shù)的兩個參數(shù)分別是 resolve 和 reject 雏掠。resolve 函數(shù)的作用是斩祭，將 Promise 對象的狀態(tài)從 pending 變?yōu)?resolved，在異步操作成功時調(diào)用乡话，并將異步操作的結(jié)果停忿，作為參數(shù)傳遞出去；reject 函數(shù)的作用是蚊伞，將 Promise 對象的狀態(tài)從從 pending 變?yōu)?rejected席赂，在異步操作失敗時調(diào)用，并將異步操作報出的錯誤时迫，作為參數(shù)傳遞出去颅停。
然后我們可以指定 then 和 catch 來分別接收 resolved 和 reject 回調(diào)。
getJSON("/posts.json")
.then(result => {···})
.catch(error => {···})
then 所指定的函數(shù)中還可以返回 Promise 對象掠拳，相當(dāng)于一個異步操作里面做了另一個異步操作癞揉。
getJSON("/post/1.json").then(function(post) {
return getJSON(post.commentURL);
}).then(function (comments) {
console.log("resolved: ", comments);
}, function (err){
console.log("rejected: ", err);
});
所以我們上面提到的網(wǎng)絡(luò)請求利用 Promise 最終可以寫成這樣：
fetch(url).then(function(response) {
return response.json()
}).then(function (json) {
//處理json
}).catch(function (err){
console.log("rejected: ", err);
});
可以看到，通過 Promise 的改造后溺欧，嵌套的異步編程變得很清晰了喊熟。
Generator
上面講了異步的一種新的解決方案 Promise,但它并不是基于協(xié)程的，只是內(nèi)部對回調(diào)函數(shù)做了封裝姐刁。下面我們會介紹一種真正基于協(xié)程的異步編程解決方案 Generator芥牌。
function* gen(x) {
var y = yield x + 2;
return y;
}
var g = gen(1);
g.next() // { value: 3, done: false }
g.next() // { value: undefined, done: true }
ES6 中，用 function* 表示協(xié)程函數(shù)聂使。調(diào)用gen(1)的時候函數(shù)不會立即執(zhí)行壁拉，而是返回一個遍歷器（提供 next 方法來遍歷）谬俄。調(diào)用 g.next() 會真正執(zhí)行函數(shù), 當(dāng)遇到 yield 的時候，主函數(shù)會出讓控制權(quán)弃理，轉(zhuǎn)而執(zhí)行 yield 后面的語句溃论。g.next() 的返回值是一個對象，value 代表 yield 后面表達(dá)式的返回值痘昌，done 代表當(dāng)前的 Generator 函數(shù)是否執(zhí)行完畢钥勋。
Generator 的執(zhí)行過程可以用下圖表示：
上面的示例 Generator 只是處理了一個同步操作，但其實 Generator 的威力是在處理異步上辆苔，下面來看一下一個 Generator 在異步情況下的處理算灸。由于異步操作的特殊性，在 yield 切換到異步操作的時候姑子，并不知道具體什么時候返回，所以在這里需要借助 Thunk 函數(shù)和遞歸來實現(xiàn)異步 Generator 的自動執(zhí)行 测僵。JS 有一個叫做 co 的庫可以讓異步 Generator 自動執(zhí)行街佑。
定義一個 Generator 函數(shù)：
var gen = function* () {
var f1 = yield readFile('/etc/fstab');
var f2 = yield readFile('/etc/shells');
console.log(f1.toString());
console.log(f2.toString());
};
利用 co 可以自動執(zhí)行異步 Generator：
var co = require('co');
co(gen);
co 會返回一個 Promise 對象，用于訂閱函數(shù)執(zhí)行完畢的回調(diào)：
co(gen).then(function (){
console.log('Generator 函數(shù)執(zhí)行完成');
});
async / await
ES2017 標(biāo)準(zhǔn)引入了 async 函數(shù)捍靠，使得異步操作更加方便沐旨。我們上面用 Generator 實現(xiàn)的讀取文件操作，轉(zhuǎn)化成 async 函數(shù)就是這樣：
const asyncReadFile = async function () {
const f1 = await readFile('/etc/fstab');
const f2 = await readFile('/etc/shells');
console.log(f1.toString());
console.log(f2.toString());
};
其實 async / await 函數(shù)就是 generator 的語法糖榨婆。把 * 號換成了 async磁携，把 yield 換成了 await。Generator 的執(zhí)行需要執(zhí)行器良风，所以才有了上面提到的 co 庫谊迄。而 async 函數(shù)自帶執(zhí)行器，所以調(diào)用它可以像調(diào)用普通函數(shù)那樣：
asyncReadFile()
下面用 async 函數(shù)改寫一個普通的網(wǎng)絡(luò)請求場景：
async function loadData() {
let response = await fetch(url)
let responseJson = await response.json()
//接下來對 json 進(jìn)行操作.....
}
await 后面一般是一個 Promise 對象烟央，也可以跟基礎(chǔ)類型统诺。如果是基礎(chǔ)類型 await 會立刻返回。
iOS
對于 iOS 來說疑俭，蘋果一直沒有推出協(xié)程的 API粮呢，iOS 的異步編程一直都是基于 Block 實現(xiàn)〕В可以用一張圖詮釋 iOS 的異步編程現(xiàn)狀：
基于Block回調(diào)的異步編程方式有以下缺點：
- 容易進(jìn)入"嵌套地獄"
- 錯誤處理復(fù)雜和冗長
- 容易忘記調(diào)用completion handler
- 條件執(zhí)行變得很困難
- 從互相獨立的調(diào)用中組合返回結(jié)果變得極其困難
- 在錯誤的線程中繼續(xù)執(zhí)行（如子線程操作UI）
而就在今年的早些時候啄寡，一篇名為「剛剛，阿里開源 iOS 協(xié)程開發(fā)框架 coobjc哩照！」的文章引爆了朋友圈挺物，阿里推出了 iOS 上的協(xié)程框架 「coobjc」。
先來看一下這個庫的架構(gòu)：
- 最底層是協(xié)程內(nèi)核飘弧，包含了棧切換的管理姻乓、協(xié)程調(diào)度器的實現(xiàn)嵌溢、協(xié)程間通信channel的實現(xiàn)等
- 中間層是基于協(xié)程的操作符的包裝，目前支持async/await蹋岩、Generator赖草、Actor等編程模型
- 最上層是對系統(tǒng)庫的協(xié)程化擴(kuò)展，目前基本上覆蓋了Foundation和UIKit的所有IO和耗時方法
以下是這個庫的核心能力：
提供了類似
C#和Javascript語言中的Async/Await編程方式支持剪个，在協(xié)程中通過調(diào)用await方法即可同步得到異步方法的執(zhí)行結(jié)果秧骑，非常適合IO、網(wǎng)絡(luò)等異步耗時調(diào)用的同步順序執(zhí)行改造扣囊。提供了類似
Kotlin中的Generator功能乎折，用于懶計算生成序列化數(shù)據(jù)，非常適合多線程可中斷的序列化數(shù)據(jù)生成和訪問侵歇。提供了
Actor Model的實現(xiàn)骂澄，基于Actor Model，開發(fā)者可以開發(fā)出更加線程安全的模塊惕虑，避免由于直接函數(shù)調(diào)用引發(fā)的各種多線程崩潰問題坟冲。提供了元組的支持，通過元組
Objective-C開發(fā)者可以享受到類似Python語言中多值返回的好處溃蔫。
還對 Foundation 的一些基礎(chǔ)庫做了擴(kuò)展：
提供了對
NSArray健提、NSDictionary等容器庫的協(xié)程化擴(kuò)展，用于解決序列化和反序列化過程中的異步調(diào)用問題伟叛。提供了對
NSData私痹、NSString、UIImage等數(shù)據(jù)對象的協(xié)程化擴(kuò)展统刮，用于解決讀寫IO過程中的異步調(diào)用問題紊遵。提供了對
NSURLConnection和NSURLSession的協(xié)程化擴(kuò)展，用于解決網(wǎng)絡(luò)異步請求過程中的異步調(diào)用問題侥蒙。提供了對
NSKeyedArchieve癞蚕、NSJSONSerialization等解析庫的擴(kuò)展，用于解決解析過程中的異步調(diào)用問題辉哥。
官方也提供了使用示例：
經(jīng)過協(xié)程改造后的網(wǎng)絡(luò)請求：
在 coobjc 中桦山，也可以找到上面提到的 ES6 中的 API：
COPromise
COPromise 的用法同 ES6 Promiese 對象的用法基本一樣：
- (void)testCoobjcPromise {
COPromise *promise = [COPromise promise:^(COPromiseFulfill _Nonnull fullfill, COPromiseReject _Nonnull reject) {
dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(3 * NSEC_PER_SEC)), dispatch_get_main_queue(), ^{
fullfill(@"異步任務(wù)結(jié)束");
});
} onQueue:dispatch_get_global_queue(0, 0)];
[promise then:^id _Nullable(id _Nullable value) {
NSLog(@"%@",value);
return nil;
}];
}
在延遲三秒之后，控制臺輸出了我們在 Promise 內(nèi)部 fullfill 回調(diào)中傳入的字符串 “異步任務(wù)結(jié)束” 醋旦。
COGenerator
COGenerator 也類似恒水，只是寫法上略有不同：
- (void)testGenerator {
COGenerator *generator = [[COGenerator alloc] initWithBlock:^{
int index = 0;
while(co_isActive()){
yield_val(@(index));
index++;
}
} onQueue:dispatch_get_global_queue(0, 0) stackSize:1024];
co_launch(^{
for(int i = 0; i < 10; i++){
int val = [[generator next] intValue];
NSLog(@"generator______value:%d",val);
}
});
}
COChan
COChan 可以理解為一個管道，管道兩邊分別是消息的發(fā)送者和接收者饲齐。原型是基于CSP并發(fā)模型钉凌。
實際上 COChan 就是一個阻塞的消息隊列。
我們可以通過以下的方法創(chuàng)建一個 COChan：
COChan *chan = [COChan chanWithBuffCount:1];
buffCount 表示該 chan 可以容納的消息數(shù)量捂人，因為在阻塞的條件下面 chan 內(nèi)部的消息會堆積御雕。
創(chuàng)建好 chan 之后矢沿，可以通過 send 和 receive 方法發(fā)送和接收 chan 的消息。send 和 receive 有兩種酸纲，一種是阻塞式的 send捣鲸、receive，一種是非阻塞式的 send_nonblock闽坡、receive_nonblock栽惶。以下是一些 send 和 receive 需要注意的點：
-
send、receive必須在協(xié)程中調(diào)用疾嗅。當(dāng)調(diào)用send或者receive外厂，會使當(dāng)前的協(xié)程掛起直到buffCount為 0 的時候恢復(fù)。也就是說代承，如果調(diào)用了send協(xié)程會一直等待直到有人調(diào)用了receive汁蝶，反過來調(diào)用receive也是一樣的道理。 -
send_nonblock论悴、receive_nonblock不會阻塞協(xié)程掖棉，所以也不需要在協(xié)程中調(diào)用。
Await
與 ES6 不同的是意荤， coobjc 的 await 方法里面只能處理 COPromise 和 COChan 類型的對象：
- (COPromise *)Promise {
COPromise *promise = [COPromise promise:^(COPromiseFulfill _Nonnull fullfill, COPromiseReject _Nonnull reject) {
dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(3 * NSEC_PER_SEC)), dispatch_get_main_queue(), ^{
fullfill(@"異步任務(wù)結(jié)束");
});
} onQueue:dispatch_get_global_queue(0, 0)];
return promise;
}
-(void)testAwait {
co_launch(^{
id result = await([self Promise]);
NSLog(@"%@", result);
});
}
在這里我們構(gòu)造了一個 COPromise 對象啊片，并在內(nèi)部模擬了一個異步操作只锻，然后我們把這個COPromise 對象傳到了 await 方法里玖像，在 await 方法的返回值里我們得到了這個 COPromise 對象的輸出。
注意： coobjc 協(xié)程上的任務(wù)都需要執(zhí)行在 co_launch 的回調(diào)中齐饮。
co_await 執(zhí)行過程：
三.優(yōu)勢
介紹完了上面幾種協(xié)程的 API捐寥，那么相對傳統(tǒng)的基于子例程的開發(fā)方式，協(xié)程到底帶來了什么優(yōu)勢祖驱？
異步操作的同步化表達(dá)
假設(shè)我們有3個異步任務(wù)需要同步執(zhí)行握恳，并且它們之間互相依賴，那么基于傳統(tǒng)的回調(diào)函數(shù)是下面這樣：
- (void)testNormalAsyncFunc {
@weakify(self);
[self asyncTask:^(NSInteger number) {
@strongify(self);
[self asyncTask:^(NSInteger number) {
@strongify(self);
NSInteger num = number + 1;
[self asyncTask:^(NSInteger number) {
@strongify(self);
NSLog(@"testNormalAsyncFunc_____%ld",(long)number);
} withNumber:num];
} withNumber:number];
} withNumber:1];
}
- (void)asyncTask:(void(^)(NSInteger number))callBack withNumber:(NSInteger)number {
dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(2 * NSEC_PER_SEC)), dispatch_get_main_queue(), ^{
if (callBack) {
callBack(number);
}
});
}
傳統(tǒng)的回調(diào)函數(shù)在代碼上閱讀不是特別友好捺僻，并且需要處理多級 Block 嵌套過程中循環(huán)引用的問題乡洼。在這里我是直接忽略調(diào)了錯誤處理，真實的環(huán)境中錯誤處理不容忽視匕坯，那么在多級回調(diào)中就會變得特別復(fù)雜束昵，這也是文章一開始說的回調(diào)地獄。
下面葛峻，我們看下經(jīng)過協(xié)程改造后的代碼：
- (void)testCORoutineAsyncFunc {
co_launch(^{
NSNumber *num = await([self promiseWithNumber:@(1)]);
// NSError *error = co_getError(); 如果有錯誤锹雏，可以這樣獲取
num = await([self promiseWithNumber:@(num.integerValue + 1)]);
num = await([self promiseWithNumber:@(num.integerValue + 1)]);
NSLog(@"testCORoutineAsyncFunc______%@",num);
});
}
- (COPromise *)promiseWithNumber:(NSNumber *)number {
COPromise *promise = [COPromise promise:^(COPromiseFulfill _Nonnull fullfill, COPromiseReject _Nonnull reject) {
dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(3 * NSEC_PER_SEC)), dispatch_get_main_queue(), ^{
fullfill(number);
// reject(error); // 如果有錯誤，回調(diào)到上層
});
} onQueue:dispatch_get_global_queue(0, 0)];
return promise;
}
可以看到术奖，異步流程非常清晰礁遵，就像寫同步代碼一樣轻绞。另外，如果發(fā)生錯誤佣耐，在 Promise 內(nèi)部會調(diào)用 reject 回調(diào)將錯誤回調(diào)出去政勃，在 await 之后可以調(diào)用 co_getError 獲取到錯誤。
對于 IO 的封裝
iOS 系統(tǒng) API 設(shè)計很不友好晰赞，絕大部分 IO稼病、跨進(jìn)程調(diào)用等耗時接口提供的都是同步方法，主線程調(diào)用會產(chǎn)生嚴(yán)重性能問題掖鱼。coobjc 封裝了基本所有的 IO API然走，讓 IO 擁有了協(xié)程調(diào)度的能力。
//從網(wǎng)絡(luò)異步加載數(shù)據(jù)
[NSURLSession sharedSession].configuration.requestCachePolicy = NSURLRequestReloadIgnoringCacheData;
NSURLSessionDownloadTask *task = [[NSURLSession sharedSession] downloadTaskWithURL:url completionHandler:
^(NSURL *location, NSURLResponse *response, NSError *error) {
if (error) {
return;
}
//在子線程解析數(shù)據(jù)戏挡，并生成圖片
dispatch_async(dispatch_get_global_queue(0, 0), ^{
NSData *data = [[NSData alloc] initWithContentsOfURL:location];
UIImage *image = [[UIImage alloc] initWithData:data];
dispatch_async(dispatch_get_main_queue(), ^{
//調(diào)度到主線程顯示圖片
imageView.image = image;
});
});
}];
協(xié)程改造后：
co_launch(^{
NSData *data = await(downloadDataFromUrl(url));
UIImage *image = await(imageFromData(data));
imageView.image = image;
});
懶加載序列
懶加載序列也是協(xié)程的一個經(jīng)典用法芍瑞。下面我們利用 COGenerator 生一個懶加載的斐波那契序列：
- (void)testFibonacciLazySequence {
COGenerator *fibonacci = [[COGenerator alloc] initWithBlock:^{
yield_val(@(1));
int cur = 1;
int next = 1;
while(co_isActive()){
yield(@(next));
int temp = cur + next;
cur = next;
next = temp;
}
} onQueue:dispatch_get_global_queue(0, 0) stackSize:1024];
co_launch(^{
for(int i = 0; i < 10; i++){
int val = [[fibonacci next] intValue];
NSLog(@"fibonacciLazySequence______value:%d",val);
}
});
}
利用 COGenerator 我們生成了一個懶加載的斐波那契數(shù)列并打印了數(shù)列的前10個元素，由于是懶加載序列褐墅，只有在調(diào)用 next 方法的時候才會真正在內(nèi)存中生成這個元素拆檬。并且，生成器的實現(xiàn)是線程安全的妥凳，因為它們都是在單線程上運行竟贯，數(shù)據(jù)在生成器中生成，然后在另一條協(xié)程上使用逝钥，期間不需要加任何鎖屑那。而使用傳統(tǒng)容器需要注意線程安全問題并且容易引發(fā) crash。
下面一張圖里詮釋了 Generator 和傳統(tǒng)容器的區(qū)別：
傳統(tǒng)的數(shù)組艘款，為了保障線程安全持际，讀寫操作必須放到一個同步隊列中。而 generator 不需要哗咆，generator 可以在一個線程生成數(shù)據(jù)蜘欲，另外一個線程消費數(shù)據(jù)。
Actor Model
race condition
大家應(yīng)該都聽說過進(jìn)程和線程晌柬，也都知道進(jìn)程和線程的概念姥份。簡單概括就是：
對操作系統(tǒng)來說，線程是最小的執(zhí)行單元年碘，進(jìn)程是最小的資源管理單元澈歉。
進(jìn)程和線程都是由操作系統(tǒng)調(diào)度和管理的，開發(fā)者并沒有權(quán)限∈⑴荩現(xiàn)在主流的操作系統(tǒng)下闷祥，線程都是搶占式的。每個線程有特定的時間片，當(dāng)線程時間片耗盡凯砍，操作系統(tǒng)就會讓線程暫停轉(zhuǎn)而去執(zhí)行其它線程箱硕。這就造成了在多線程下，線程與線程之間會出現(xiàn) race condition悟衩。舉一個最常見的 i++ 的例子：
i++ 操作在底層其實是有三步：
- 寄存器從變量讀取 i 的值
- 在寄存器中對 i 進(jìn)行 +1 操作
- 把加完后的值寫回變量
在多線程情況下剧罩，就有可能出現(xiàn)如圖所示的情況。
為了避免這個問題座泳，我們需要對出現(xiàn) race condition 的資源加鎖惠昔。但是鎖會帶來以下問題：
- 鎖對于系統(tǒng)資源的占用。
- 涉及到線程阻塞狀態(tài)和可運行狀態(tài)之間的切換挑势。
- 涉及到線程上下文的切換镇防。
以上涉及到的任何一點，都是非常耗費性能的操作潮饱。
Actor
為了避免頻繁的線程加鎖和線程切換来氧，我們引入了一種新的并發(fā)模型：
Actor Model 是一種替代多線程的并發(fā)解決方案。傳統(tǒng)的并發(fā)解決方案是一種共享數(shù)據(jù)方式香拉，使用共享數(shù)據(jù)的并發(fā)編程面臨的最大問題是數(shù)據(jù)條件競爭啦扬，處理各種鎖的問題是讓人十分頭疼的。
首先我們來看一下 Actor 的結(jié)構(gòu)：
Actor 模型的理念是：萬物皆為 Actor凫碌。Actor 與 Actor 之間是通過消息通信扑毡，并不會直接共享每一個資源。Actor 內(nèi)部有一個 Mailbox 盛险，可以理解為一個消息隊列瞄摊。所有消息都會先發(fā)送到 Mailbox。Actor 內(nèi)部管理著自身的 State枉层，這個 State 只有 Actor 自己可以訪問泉褐。
讓我們來看一下 Actor 和傳統(tǒng) OOP 的對比：
示例
計數(shù)器
假設(shè)我們現(xiàn)在需要實現(xiàn)一個計數(shù)器赐写，傳統(tǒng)的實現(xiàn)應(yīng)該是這樣：
如果不加鎖鸟蜡，我們使用多線程對計數(shù)器進(jìn)行累加操作后，是達(dá)不到我們想得到的目標(biāo)值的挺邀。原因就是上面提到的 race condition揉忘。
經(jīng) Actor 改造后：
COActor *countActor = co_actor_onqueue(get_test_queue(), ^(COActorChan *channel) {
int count = 0;
for(COActorMessage *message in channel){
if([[message stringType] isEqualToString:@"inc"]){
count++;
}
else if([[message stringType] isEqualToString:@"get"]){
message.complete(@(count));
}
}
});
我們使用 Actor 來測試一下多線程下的計數(shù)器累加操作：
NSOperationQueue *queue = [[NSOperationQueue alloc] init];
queue.maxConcurrentOperationCount = 15;
for (int i = 0;i < 10000;i++) {
NSBlockOperation *operation = [NSBlockOperation blockOperationWithBlock:^{
[countActor sendMessage:@"inc"];
}];
[queue addOperation:operation];
}
[queue waitUntilAllOperationsAreFinished];
co_launch(^{
int currentCount = [await([countActor sendMessage:@"get"]) intValue];
NSLog(@"count: %d", currentCount);
});
可以看到最后輸出的值是我們期望的目標(biāo)值，可見 Actor 在多線程下是線程安全的端铛。
求素數(shù)
假設(shè)現(xiàn)在有一個任務(wù)，利用多線程找100000以內(nèi)的素數(shù)的個數(shù)，如果以傳統(tǒng)的共享數(shù)據(jù)的方式锈候，我們需要兩個變量绑洛，一個記錄當(dāng)前找到了第幾，一個記錄當(dāng)前素數(shù)的個數(shù)。因為是多線程哗总，我們在操作這兩個變量的時候需要對它們加鎖几颜。加鎖的壞處大家也很清楚，首先是加鎖的開銷是很大的讯屈，而且如果不小心還會出現(xiàn)死鎖蛋哭。
下面我們來使用基于分布式的 Actor Model 來解決這個問題。
如果用 Actor 模型實現(xiàn)統(tǒng)計素數(shù)個數(shù)涮母，那么我們需要1個 Actor 做原料的分發(fā)谆趾，就是提供要處理的整數(shù)，然后3個 Actor 加工叛本，每次從分發(fā) Actor 那里拿一個整數(shù)進(jìn)行加工沪蓬，最終把加工出來的半成品發(fā)給組裝 Actor，組裝 Actor 把3個加工 Actor 的結(jié)果匯總輸出来候。
下面是一個分布式的結(jié)構(gòu)：
COActor 就是 coobjc 為我們提供的 Actor 封裝類怜跑，下面我們用 COActor 實現(xiàn)上面的分布式結(jié)構(gòu):
#import "Actor.h"
#import <coobjc/coobjc.h>
@interface Actor ()
@property (nonatomic, strong) COActor *assembler;
@property (nonatomic, strong) COActor *dispatcher;
@property (nonatomic, strong) NSMutableArray<COActor *> *processers;
@property (nonatomic, assign) NSTimeInterval startTime;
@end
@implementation Actor
- (instancetype)init
{
self = [super init];
if (self) {
_processers = [NSMutableArray arrayWithCapacity:0];
[self setupProcessers];
__weak Actor *weakSelf = self;
_assembler = co_actor_onqueue(dispatch_get_global_queue(0, 0), ^(COActorChan * _Nonnull chan) {
__block int numCount = 0;
for (COActorMessage *message in chan) {
if ([message.stringType isEqualToString:@"add"]) {
numCount ++;
} else if ([message.stringType isEqualToString:@"stop"]) {
NSTimeInterval costTime = CFAbsoluteTimeGetCurrent() - self.startTime;
NSLog(@"素數(shù)的個數(shù)是______%d 消耗的時間_________%f", numCount, costTime);
}
}
});
_dispatcher = co_actor_onqueue(dispatch_get_global_queue(0, 0), ^(COActorChan * _Nonnull chan) {
for (COActorMessage *message in chan) {
if ([message.stringType isEqualToString:@"start"]) {
weakSelf.startTime = CFAbsoluteTimeGetCurrent();
__block int number = 0;
while (number <= 100000) {
[weakSelf.processers enumerateObjectsUsingBlock:^(COActor * _Nonnull obj, NSUInteger idx, BOOL * _Nonnull stop) {
dispatch_async(dispatch_get_global_queue(0, 0), ^{
[obj sendMessage:@(number++)];
});
}];
}
}
}
});
}
return self;
}
- (void)setupProcessers {
__weak Actor *weakSelf = self;
for (int i = 0;i < 4;i++) {
COActor *processer = co_actor_onqueue(dispatch_get_global_queue(0, 0), ^(COActorChan * _Nonnull chan) {
for (COActorMessage *message in chan) {
NSInteger currentNum = message.uintType;
if ([weakSelf isPrimeNumber:currentNum]) {
[[weakSelf assembler] sendMessage:@"add"];
}
if (currentNum == 10000) {
NSLog(@"processer%d stop",i+1);
[weakSelf.assembler sendMessage:@"stop"];
}
}
});
[self.processers addObject:processer];
}
}
- (BOOL)isPrimeNumber:(NSInteger)number {
BOOL flag = YES;
for (int i = 2;i < number;i++) {
if (number % i == 0) {
flag = NO;
break;
}
}
return flag;
}
- (void)startTask {
[self.dispatcher sendMessage:@"start"];
}
@end
dispatcher 作為自然數(shù)的生產(chǎn)者，產(chǎn)生 10000 以內(nèi)的自然數(shù)吠勘。
processer 作為加工者性芬，來處理自然數(shù)是否是素數(shù)的工作。如果是素數(shù)剧防，就會發(fā)消息到 assembler植锉。
assembler 作為組裝者，有一個內(nèi)部狀態(tài) numCount 記錄目前素數(shù)的個數(shù)峭拘。當(dāng)收到加工者的消息后俊庇，會把 numCount 自增。
可以看到鸡挠，基于分布式的 Actor Model辉饱，很好的避免了傳統(tǒng)并發(fā)模型下的共享資源，沒有任何的鎖操作拣展，并且對每一個 Actor 都進(jìn)行了明確的功能劃分彭沼。
四.原理
剛才開篇有提到，協(xié)程與普通函數(shù)調(diào)用的區(qū)別就是备埃，協(xié)程可以隨時中斷協(xié)程的執(zhí)行姓惑，跳轉(zhuǎn)到新的協(xié)程里面去，并且還能在需要的時候恢復(fù)按脚。而普通的函數(shù)調(diào)用是沒法作出暫停操作的于毙。所以實現(xiàn)協(xié)程的關(guān)鍵就在于把當(dāng)前調(diào)用棧上的狀態(tài)都保存下來，然后再能從緩存中恢復(fù)辅搬。
基本上市面上實現(xiàn)這種操作的方法有以下五種：
- 利用 glibc 的
ucontext組件(云風(fēng)的庫)唯沮。 - 使用匯編代碼來切換上下文(實現(xiàn)c協(xié)程)，原理同
ucontext。 - 利用C語言語法
switch-case的奇淫技巧來實現(xiàn)（Protothreads)介蛉。 - 利用了 C 語言的
setjmp和longjmp夯缺。 - 利用編譯器支持語法糖。
這里我主要介紹第一種甘耿，其它的大家感興趣可以自行 Google踊兜。
這里順便說一下第一種方法在 iOS 上是已經(jīng)被廢棄的（可能是因為效率不高）。第三種和第四種只是能過做到跳轉(zhuǎn)佳恬，但是沒法保存調(diào)用棧上的狀態(tài)捏境，看起來基本上不能算是實現(xiàn)了協(xié)程，只能算做做demo毁葱，第五種除非官方支持垫言，否則自行改寫編譯器通用性很差。所以阿里爸爸的庫其實是使用第二種方法實現(xiàn)的倾剿。
uncontext
一.Hello world
uncontext 是一個 c 的協(xié)程庫筷频，底層也是通過匯編來實現(xiàn)的。首先我們來看一個最簡單的例子：
#include <stdio.h>
#include <ucontext.h>
#include <unistd.h>
int main(int argc, const char *argv[]){
ucontext_t context;
getcontext(&context);
printf("Hello world");
sleep(1);
setcontext(&context);
return 0;
}
運行這段代碼前痘，打印如下：
Hello world
Hello world
Hello world
Hello world
Hello world
Hello world
Hello world
...............
如果我們不終止程序凛捏，會一直打印 "Hello world"。
在這里我們初步可以看到芹缔，getcontext 保存了當(dāng)前函數(shù)的執(zhí)行上下文坯癣，當(dāng)程序執(zhí)行到 setcontext 的時候又會回到 getcontext 的那一行繼續(xù)執(zhí)行。
二.uncontext 深入
這里我們需要關(guān)注 <ucontext.h> 中定義的兩個結(jié)構(gòu)體類型 mcontext_t 和 ucontext_t最欠，和四個函數(shù) getcontext()示罗、setcontext()、makecontext()芝硬、swapcontext()蚜点。
首先來看下 uncontext 的結(jié)構(gòu)：
typedef struct ucontext {
struct ucontext *uc_link;
sigset_t uc_sigmask;
stack_t uc_stack;
mcontext_t uc_mcontext;
...
} ucontext_t;
- uc_link：指向當(dāng)前上下文運行終止時會回復(fù)的上下文
- uc_sigmask：上下文要阻塞的信號集合
- uc_stack：上下文所使用的棧空間
- uc_mcontext：其中 mcontext_t 類型與機(jī)器相關(guān)的類型拌阴。這個字段是機(jī)器特定的保護(hù)上下文的表示绍绘，包括協(xié)程的機(jī)器寄存器
下面說一下四個函數(shù)：
int getcontext(ucontext_t *ucp);：保存當(dāng)前的 context 到 ucp 結(jié)構(gòu)體中。int setcontext(const ucontext_t *ucp);：設(shè)置當(dāng)前的上下文為 ucp 皮官。這里會判斷如果上下文是通過getcontext()獲取的脯倒，程序會直接執(zhí)行這個上下文实辑。如果上下文是通過makecontext()獲取的捺氢，程序會執(zhí)行調(diào)用makecontext()第二個參數(shù)(func)所指向的函數(shù)。如果 func 函數(shù)返回剪撬，則會執(zhí)行makecontext()第一個參數(shù)ucp.uc_link所指向的上下文摄乒。如果ucp.uc_link == NULL則程序退出。void makecontext(ucontext_t *ucp, void (*func)(), int argc, ...);：調(diào)用makecontext()之前需要先調(diào)用getcontext()生成上下文 ucp。makecontext()會給該 ucp 指定一個椻捎樱空間ucp->uc_stack斋否。最重要的是設(shè)置后繼上下文ucp->uc_link，這里確保了我們可以在協(xié)程間進(jìn)行切換拭荤。然后這里會傳入一個函數(shù)指針茵臭，當(dāng)后面我們調(diào)用setcontext()和swapcontext()的時候會直接調(diào)用該函數(shù)。具體的在上面setcontext()方法講解的時候有講到舅世。int swapcontext(ucontext_t *oucp, ucontext_t *ucp);：保存當(dāng)前的上下文當(dāng) oucp 中旦委，然后激活 ucp 所指向的上下文。
下面我們結(jié)合上面給的函數(shù)雏亚，來看一個簡單的例子：
#include <ucontext.h>
#include <stdio.h>
void func1(void * arg)
{
printf("func1");
}
void context_test()
{
char stack[1024*128];
ucontext_t child,main;
print("context_test");
getcontext(&child); //獲取當(dāng)前上下文
child.uc_stack.ss_sp = stack;//指定椨酰空間
child.uc_stack.ss_size = sizeof(stack);//指定棧空間大小
child.uc_link = &main;//設(shè)置后繼上下文
makecontext(&child,(void (*)(void))func1,0);//修改上下文指向func1函數(shù)
swapcontext(&main,&child);//切換到child上下文罢低，保存當(dāng)前上下文到main
printf("main");//如果設(shè)置了后繼上下文查辩，func1函數(shù)指向完后會返回此處
}
int main()
{
context_test();
return 0;
}
執(zhí)行上面代碼，會打印出
contest_test
func1
main
現(xiàn)在來分析一下這段代碼：按照函數(shù)執(zhí)行順序网持，首先會打印 "context_test" 宜岛。然后這里用 getcontext() 獲取了當(dāng)前上下文并設(shè)置了棧空間和后繼上下文功舀。注意這里 main 結(jié)構(gòu)體并沒有真正被賦值谬返。隨后調(diào)用了 makecontext() 并傳入了一個指向 func1 的函數(shù)指針，那么在下面調(diào)用 swapcontext() 的時候就會調(diào)用這個函數(shù)指針?biāo)赶虻暮瘮?shù)日杈，也就是 func1遣铝。swapcontext() 還有一個功能就是把當(dāng)前上下文賦值給第一個參數(shù)，所以現(xiàn)在 main 里面存儲了當(dāng)前的上下文莉擒。最后當(dāng) func1 執(zhí)行完之后酿炸，由于之前 child.uc_link 設(shè)置的是 main，所以會切換到 main 的上下文繼續(xù)執(zhí)行涨冀，所以我們看到了最后打印的 "main"填硕。
現(xiàn)在我們稍微改動一下代碼，把之前的 uc_link 賦值改為 child.uc_link = NULL,然后我們再次運行代碼鹿鳖，可以看到打印變成了：
contest_test
func1
因為沒有設(shè)置后繼上下文扁眯，所以程序在執(zhí)行完 func1 之后就直接結(jié)束了。
三.如何實現(xiàn)協(xié)程翅帜？
上面的例子只是簡單的實現(xiàn)了從 主協(xié)程->協(xié)程1->主協(xié)程 這樣一條路徑姻檀，這樣的路徑普通函數(shù)調(diào)用也能實現(xiàn)。下面我們會介紹中斷協(xié)程和恢復(fù)協(xié)程如何實現(xiàn)涝滴。
實現(xiàn)協(xié)程我們首先需要定義兩個結(jié)構(gòu)绣版，一個是協(xié)程胶台，一個是協(xié)程的調(diào)度器。
我這里的定義參考了 coobjc 和 uthread 的作者杂抽。
協(xié)程的結(jié)構(gòu)體定義如下：
typedef void (*entryFunc)(void *arg);
typedef struct coroutine
{
ucontext_t ctx; //當(dāng)前協(xié)程的上下文诈唬，用于后繼協(xié)程的存儲
entryFunc func; //協(xié)程需要執(zhí)行的函數(shù)
void *arg; // 函數(shù)的執(zhí)行參數(shù)
enum ThreadState state; //協(xié)程的運行狀態(tài)，包括 FREE缩麸、RUNNABLE铸磅、RUNING、SUSPEND 四種杭朱。
char stack[DEFAULT_STACK_SZIE]; //椨奁ǎ空間
struct coroutine_scheduler *scheduler; // 協(xié)程的調(diào)度器.
};
調(diào)度器的定義如下：
typedef std::vector<uthread_t> Thread_vector; //使用 c++數(shù)組來裝載協(xié)程對象
typedef struct coroutine_scheduler
{
ucontext_t main; // main的上下文
int running_thread; // 當(dāng)前正在運行的協(xié)程編號，如果沒有返回 -1
Thread_vector threads; //協(xié)程數(shù)組
schedule_t():running_thread(-1){}
}
還需要實現(xiàn)4個比較關(guān)鍵的方法痕檬。
int uthread_create(schedule_t &schedule,Fun func,void *arg)：創(chuàng)建協(xié)程
int uthread_create(schedule_t &schedule,Fun func,void *arg)
{
int id = 0;
for(id = 0; id < schedule.max_index; ++id ){
if(schedule.threads[id].state == FREE){
break;
}
}
if (id == schedule.max_index) {
schedule.max_index++;
}
// 加入到協(xié)程隊列
uthread_t *t = &(schedule.threads[id]);
//初始化協(xié)程結(jié)構(gòu)體
t->state = RUNNABLE;
t->func = func;
t->arg = arg;
//設(shè)置協(xié)程的上下文
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_body),1,&schedule);
swapcontext(&(schedule.main), &(t->ctx));
return id;
}
void uthread_yield(schedule_t &schedule)：掛起協(xié)程
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 uthread_resume(schedule_t &schedule , int id)：恢復(fù)協(xié)程
void uthread_resume(schedule_t &schedule , int id)
{
if(id < 0 || id >= schedule.max_index){
return;
}
uthread_t *t = &(schedule.threads[id]);
if (t->state == SUSPEND) {
swapcontext(&(schedule.main),&(t->ctx));
}
}
int schedule_finished(const schedule_t &schedule)：所有協(xié)程執(zhí)行完畢
int schedule_finished(const schedule_t &schedule)
{
if (schedule.running_thread != -1){
return 0;
}else{
for(int i = 0; i < schedule.max_index; ++i){
if(schedule.threads[i].state != FREE){
return 0;
}
}
}
return 1;
}
接下來用 uthread 里提供的例子來驗證一下協(xié)程庫的運行：
void func1(void * arg)
{
puts("1");
puts("11");
}
void func2(void * arg)
{
puts("22");
puts("22");
uthread_yield(*(schedule_t *)arg);
puts("22");
puts("22");
}
void func3(void *arg)
{
puts("3333");
puts("3333");
uthread_yield(*(schedule_t *)arg);
puts("3333");
puts("3333");
}
void context_test()
{
char stack[1024*128];
ucontext_t uc1,ucmain;
getcontext(&uc1);
uc1.uc_stack.ss_sp = stack;
uc1.uc_stack.ss_size = 1024*128;
uc1.uc_stack.ss_flags = 0;
uc1.uc_link = &ucmain;
makecontext(&uc1,(void (*)(void))func1,0);
swapcontext(&ucmain,&uc1);
puts("main");
}
void schedule_test()
{
schedule_t s;
int id1 = uthread_create(s,func3,&s);
int id2 = uthread_create(s,func2,&s);
while(!schedule_finished(s)){
uthread_resume(s,id2);
uthread_resume(s,id1);
}
puts("main over");
}
int main()
{
// 執(zhí)行 主協(xié)程->子協(xié)程->主協(xié)程
context_test();
//主線程和子協(xié)程間任意切換
schedule_test();
return 0;
}
上面的 Demo 可以用下面的流程圖來表示：
可以看到霎槐，在自己封裝的協(xié)程庫中，實現(xiàn)了主協(xié)程和子協(xié)程間的任意切換梦谜。并且切換之前都能保存當(dāng)前的上下文丘跌，并不會從頭開始執(zhí)行。在這里其實就體現(xiàn)了文章一開始引用的維基百科中關(guān)于協(xié)程的介紹：「協(xié)程可以通過 yield 中斷執(zhí)行唁桩，轉(zhuǎn)而執(zhí)行別的協(xié)程闭树。在這種轉(zhuǎn)換過程中不存在調(diào)用者與被調(diào)用者的關(guān)系」，這是函數(shù)調(diào)用所不能實現(xiàn)的荒澡。
總結(jié)
協(xié)程本身沒有特別適用的場景，但是當(dāng)搭配上多線程之后单山，協(xié)程的光芒漸漸顯露出來。我們可以這么總結(jié)一下：協(xié)程的線程可以讓我們的程序并發(fā)的跑昼接，協(xié)程可以讓并發(fā)程序跑得看起來更美好
不管是異步代碼同步化悴晰，還是并發(fā)代碼簡潔化，協(xié)程的出現(xiàn)其實是為代碼從計算機(jī)向人類思維的貼近提供了可能漂辐。
參考資料
剛剛棕硫，阿里開源 iOS 協(xié)程開發(fā)框架 coobjc髓涯！
基于協(xié)程的編程方式在移動端研發(fā)的思考及最佳實踐
iOS協(xié)程coobjc的設(shè)計篇-棧切換
coobjc
