即學(xué)即用Kotlin - 協(xié)程

上周在內(nèi)部分享會(huì)上大佬同事分享了關(guān)于 Kotlin 協(xié)程的知識(shí)，之前有看過(guò) Kotlin 協(xié)程的一些知識(shí)，以為自己還挺了解協(xié)程的，結(jié)果...在這一次分享中，發(fā)現(xiàn) Flow 和 Channel 這一塊兒知識(shí)是自己不怎么了解的，本文也將著重和大家聊一聊這一塊兒的內(nèi)容，協(xié)程部分將分為三篇，本文是第一篇：

“

《即學(xué)即用Kotlin - 協(xié)程》
《抽絲剝繭Kotlin - 協(xié)程基礎(chǔ)篇》
《抽絲剝繭Kotlin - 協(xié)程Flow篇》

目錄

一、基礎(chǔ)

1. 概念

相信大家或多或少的都了解過(guò)，協(xié)程是什么，官網(wǎng)上這么說(shuō)：

“

Essentially, coroutines are light-weight threads.

協(xié)程是輕量級(jí)的線程，為什么是輕量的？可以先告訴大家結(jié)論，因?yàn)樗?strong style="color: rgb(71, 193, 168);">基于線程池API，所以在處理并發(fā)任務(wù)這件事上它真的游刃有余。

有可能有的同學(xué)問(wèn)了，既然它基于線程池，那我直接使用線程池或者使用 Android 中其他的異步任務(wù)解決方式，比如 Handler、RxJava等，不更好嗎？

協(xié)程可以使用阻塞的方式寫(xiě)出非阻塞式的代碼，解決并發(fā)中常見(jiàn)的回調(diào)地獄，這是其最大的優(yōu)點(diǎn)，后面介紹。

2. 使用

GlobalScope.launch(Dispatchers.Main)?{
????val?res?=?getResult(2)
????mNumTv.text?=?res.toString()
}

啟動(dòng)協(xié)程的代碼就是如此的簡(jiǎn)單。上面的代碼中可以分為三部分，分別是 GlobalScope、Dispatcher 和 launch，他們分別對(duì)應(yīng)著協(xié)程的作用域、調(diào)度器和協(xié)程構(gòu)建器，我們挨個(gè)兒介紹。

協(xié)程作用域

協(xié)程的作用域有三種，他們分別是：

• runBlocking：頂層函數(shù)，它和 coroutineScope 不一樣，它會(huì)阻塞當(dāng)前線程來(lái)等待，所以這個(gè)方法在業(yè)務(wù)中并不適用 。
• GlobalScope：全局協(xié)程作用域，可以在整個(gè)應(yīng)用的聲明周期中操作，且不能取消，所以仍不適用于業(yè)務(wù)開(kāi)發(fā)。
• 自定義作用域：自定義協(xié)程的作用域，不會(huì)造成內(nèi)存泄漏。

顯然，我們不能在 Activity 中調(diào)用 GlobalScope，這樣可能會(huì)造成內(nèi)存泄漏，看一下如何自定義作用域，具體的步驟我在注釋中已給出：

class?MainActivity?:?AppCompatActivity()?{
????//?1.?創(chuàng)建一個(gè)?MainScope
????val?scope?=?MainScope()

????override?fun?onCreate(savedInstanceState:?Bundle?)?{
????????super.onCreate(savedInstanceState)
????????setContentView(R.layout.activity_main)
????????
????????// 2. 啟動(dòng)協(xié)程
????????scope.launch(Dispatchers.Unconfined)?{
????????????val?one?=?getResult(20)
????????????val?two?=?getResult(40)
????????????mNumTv.text?=?(one?+?two).toString()
????????}
????}

????//?3.?銷(xiāo)毀的時(shí)候釋放
????override?fun?onDestroy()?{
????????super.onDestroy()

????????scope.cancel()
????}

????private?suspend?fun?getResult(num:?Int):?Int?{
????????delay(5000)
????????return?num?*?num
????}
}

調(diào)度器

調(diào)度器的作用是將協(xié)程限制在特定的線程執(zhí)行。主要的調(diào)度器類(lèi)型有：

• Dispatchers.Main：指定執(zhí)行的線程是主線程，如上面的代碼。
• Dispatchers.IO：指定執(zhí)行的線程是 IO 線程。
• Dispatchers.Default：默認(rèn)的調(diào)度器，適合執(zhí)行 CPU 密集性的任務(wù)。
• Dispatchers.Unconfined：非限制的調(diào)度器，指定的線程可能會(huì)隨著掛起的函數(shù)的發(fā)生變化。

什么是掛起？我們就以九心吃飯為例，如果到公司對(duì)面的廣場(chǎng)吃飯，九心得經(jīng)過(guò)：

• 走到廣場(chǎng) 10min ?> 點(diǎn)餐 5min > 等待上餐 10min > 就餐 30min > 回來(lái) 10 min

如果九心點(diǎn)廣場(chǎng)的外賣(mài)呢？

• 九心：下單 5min > 等待(等待的時(shí)候可以工作) 30min > 就餐 30min
• 外賣(mài)騎手：到店 > 取餐 > 送外賣(mài)

從九心吃飯的例子可以看出，如果點(diǎn)了外賣(mài)，九心花費(fèi)的時(shí)間較少了，可以空閑出更多的時(shí)間做自己的事。再仔細(xì)分析一下，其實(shí)從公司到廣場(chǎng)和等待取餐這個(gè)過(guò)程并沒(méi)有省去，只是九心把這個(gè)過(guò)程交給了外賣(mài)員。

協(xié)程的原理跟九心點(diǎn)外賣(mài)的原理是一致的，耗時(shí)阻塞的操作并沒(méi)有減少，只是交給了其他線程：

launch

launch 的作用從它的名稱(chēng)就可以看的出來(lái)，啟動(dòng)一個(gè)新的協(xié)程，它返回的是一個(gè) Job對(duì)象，我們可以調(diào)用 Job#cancel() 取消這個(gè)協(xié)程。

除了 launch，還有一個(gè)方法跟它很像，就是 async，它的作用是創(chuàng)建一個(gè)協(xié)程，之后返回一個(gè) Deferred對(duì)象，我們可以調(diào)用 Deferred#await()去獲取返回的值，有點(diǎn)類(lèi)似于 Java 中的 ? Future，稍微改一下上面的代碼：

class?MainActivity?:?AppCompatActivity()?{
????//?1.?創(chuàng)建一個(gè)?MainScope
????val?scope?=?MainScope()

????override?fun?onCreate(savedInstanceState:?Bundle?)?{
????????super.onCreate(savedInstanceState)
????????setContentView(R.layout.activity_main)

????????// 2. 啟動(dòng)協(xié)程
????????scope.launch(Dispatchers.Unconfined)?{
????????????val?one?=?async?{?getResult(20)?}
????????????val?two?=?async?{?getResult(40)?}
????????????mNumTv.text?=?(one.await()?+?two.await()).toString()
????????}
????}

????//?3.?銷(xiāo)毀的時(shí)候釋放
????override?fun?onDestroy()?{
????????super.onDestroy()

????????scope.cancel()
????}

????private?suspend?fun?getResult(num:?Int):?Int?{
????????delay(5000)
????????return?num?*?num
????}
}

與修改前的代碼相比，async 能夠并發(fā)執(zhí)行任務(wù)，執(zhí)行任務(wù)的時(shí)間也因此縮短了一半。

除了上述的并發(fā)執(zhí)行任務(wù)，async 還可以對(duì)它的 start 入?yún)⒃O(shè)置成懶加載

val?one?=?async(start?=?CoroutineStart.LAZY)?{?getResult(20)?}

這樣系統(tǒng)就可以在調(diào)用它的時(shí)候再為它分配資源了。

suspend

suspend 是修飾函數(shù)的關(guān)鍵字，意思是當(dāng)前的函數(shù)是可以掛起的，但是它僅僅起著提醒的作用，比如，當(dāng)我們的函數(shù)中沒(méi)有需要掛起的操作的時(shí)候，編譯器回給我們提醒 Redudant suspend modifier，意思是當(dāng)前的 suspend 是沒(méi)有必要的，可以把它刪除。

那我們什么時(shí)候需要使用掛起函數(shù)呢？常見(jiàn)的場(chǎng)景有：

• 耗時(shí)操作：使用 withContext 切換到指定的 IO 線程去進(jìn)行網(wǎng)絡(luò)或者數(shù)據(jù)庫(kù)請(qǐng)求。
• 等待操作：使用delay方法去等待某個(gè)事件。

withContext ?的代碼：

private?suspend?fun?getResult(num:?Int):?Int?{
????return?withContext(Dispatchers.IO)?{
????????num?*?num
????}
}

delay 的代碼：

private?suspend?fun?getResult(num:?Int):?Int?{
????delay(5000)
????return?num?*?num
}

結(jié)合 Android Jetpack

在介紹自定義協(xié)程作用域的時(shí)候，我們需要主動(dòng)在 Activity 或者 Fragment 中的 onDestroy 方法中調(diào)用 job.cancel()，忘記處理可能是程序員經(jīng)常會(huì)犯的錯(cuò)誤，如何避免呢？

Google 總是能夠解決程序員的痛點(diǎn)，在 Android Jetpack 中的 lifecycle、LiveData 和 ViewModel 已經(jīng)集成了快速使用協(xié)程的方法，如果我們已經(jīng)引入了 Android Jetpack，可以引入依賴：

????dependencies?{
????????def?lifecycle_version?=?"2.2.0"
????????
????????//?ViewModel
????????implementation?"androidx.lifecycle:lifecycle-viewmodel-ktx:$lifecycle_version"
????????//?LiveData
????????implementation?"androidx.lifecycle:lifecycle-livedata-ktx:$lifecycle_version"
????????//?Lifecycles?only?(without?ViewModel?or?LiveData)
????????implementation?"androidx.lifecycle:lifecycle-runtime-ktx:$lifecycle_version"
????}

使用可以結(jié)合具體的場(chǎng)景，比如結(jié)合 Lifecycle，需要使用 lifecycleScope 協(xié)程作用域：

lifecycleScope.launch?{
????//?代表當(dāng)前生命周期處于?Resumed?的時(shí)候才會(huì)執(zhí)行(選擇性使用)
????whenResumed?{?
????????//?...?具體的協(xié)程代碼
????}
}

即使你不使用 Android Jetpack 組件，由于 Lifecycles 在很早之前就內(nèi)置在 Android 系統(tǒng)的代碼中，所以你仍然可以僅僅引入 Lifecycle 的協(xié)程擴(kuò)展庫(kù)，因?yàn)樗鼤?huì)幫助你很好的處理 Activity 或者 Fragment 的生命周期。

引入 Android Jetpack 協(xié)程擴(kuò)展庫(kù)官方文檔：點(diǎn)我打開(kāi)

二、流

長(zhǎng)期以來(lái)，在 Android 中響應(yīng)式編程的首選方案是 RxJava，我們今天就來(lái)了解一下 Kotlin中的響應(yīng)式編程 Flow。如果你能熟練使用 RxJava，那你肯定能快速上手 Flow。

曾經(jīng)我在《即學(xué)即用Android Jetpack - ViewModel & LiveData》一文中說(shuō)過(guò)，LiveData 的使用類(lèi)似于 RxJava，現(xiàn)在我收回這句話，事實(shí)上，LiveData 更加簡(jiǎn)單和純粹，它建立單一的生產(chǎn)消費(fèi)模型，F(xiàn)low 才是類(lèi)似于 RxJava 的存在。

1. 基礎(chǔ)

先上一段代碼：

lifecycleScope.launch?{
????//?創(chuàng)建一個(gè)協(xié)程?Flow
????createFlow()
????????.collect?{num->
????????????//?具體的消費(fèi)處理
????????????//?...
????????}
????}
}

我在 createFlow 這個(gè)方法中，返回了 Flow 的對(duì)象，所以我們可以這樣對(duì)比。

創(chuàng)建 Flow 對(duì)象

我們暫不考慮 RxJava中的背壓和非背壓，直接先將 Flow 對(duì)標(biāo) RxJava 中的 Observable。

和 RxJava 一樣，在創(chuàng)建 Flow 對(duì)象的時(shí)候我們也需要調(diào)用 emit 方法發(fā)射數(shù)據(jù)：

fun?createFlow():?Flow?=?flow?{
????for?(i?in?1..10)
????????emit(i)
}

一直調(diào)用 emit 可能不便捷，因?yàn)?RxJava 提供了 Observable.just() 這類(lèi)的操作符，顯然，F(xiàn)low 也為我們提供了快速創(chuàng)建操作：

• flowof(vararg elements: T)：幫助可變數(shù)組生成 Flow 實(shí)例
• 擴(kuò)展函數(shù) .asFlow()：面向數(shù)組、列表等集合

比如可以使用 (1..10).asFlow() 代替上述的 Flow 對(duì)象的創(chuàng)建。

消費(fèi)數(shù)據(jù)

collect 方法和 RxJava 中的 subscribe 方法一樣，都是用來(lái)消費(fèi)數(shù)據(jù)的。

除了簡(jiǎn)單的用法外，這里有兩個(gè)問(wèn)題得注意一下：

• collect 函數(shù)是一個(gè) suspend 方法，所以它必須發(fā)生在協(xié)程或者帶有 suspend 的方法里面，這也是我為什么在一開(kāi)始的時(shí)候啟動(dòng)了 lifecycleScope.launch。
• lifecycleScope 是我使用的 Lifecycle 的協(xié)程擴(kuò)展庫(kù)當(dāng)中的，你可以替換成自定義的協(xié)程作用域。

2. 線程切換

我們學(xué)習(xí) RxJava 的時(shí)候，大佬們都會(huì)說(shuō)，RxJava 牛逼，牛逼在哪兒呢？

切換線程，同樣的，F(xiàn)low 的協(xié)程切換也很牛逼。Flow 是這么切換協(xié)程的：

lifecycleScope.launch?{
????//?創(chuàng)建一個(gè)協(xié)程?Flow
????createFlow()
????????//?將數(shù)據(jù)發(fā)射的操作放到?IO?線程中的協(xié)程
????????.flowOn(Dispatchers.IO)
????????.collect?{?num?->
????????????//?具體的消費(fèi)處理
????????????//?...
????????}
????}
}

和 RxJava 對(duì)比：

改變數(shù)據(jù)發(fā)射的線程

flowOn 使用的參數(shù)是協(xié)程對(duì)應(yīng)的調(diào)度器，它實(shí)質(zhì)改變的是協(xié)程對(duì)應(yīng)的線程。

改變消費(fèi)數(shù)據(jù)的線程

我在上面的表格中并沒(méi)有寫(xiě)到在 Flow 中如何改變消費(fèi)線程，并不意味著 Flow 不可以指定消費(fèi)線程？

Flow 的消費(fèi)線程在我們啟動(dòng)協(xié)程指定調(diào)度器的時(shí)候就確認(rèn)好了，對(duì)應(yīng)著啟動(dòng)協(xié)程的調(diào)度器。比如在上面的代碼中 lifecycleScope 啟動(dòng)的調(diào)度器是 Dispatchers.Main，那么 collect 方法就消費(fèi)在主線程。

3. 異常和完成

異常捕獲

Flow 中的 catch 對(duì)應(yīng)著 RxJava 中的 onError，catch 操作：

lifecycleScope.launch?{
????flow?{
????????//...
????}.catch?{e->

????}.collect(

????)
}

除此以外，你可以使用聲明式捕獲 try { } catch (e: Throwable) { } 去捕獲異常，不過(guò) catch 本質(zhì)上是一個(gè)擴(kuò)展方法，它是對(duì)聲明式捕獲的封裝。

完成

Flow 中的 onCompletion 對(duì)應(yīng)這 RxJava 中的 onComplete 回調(diào)，onCompletion操作：

lifecycleScope.launch?{
????createFlow()
????????.onCompletion?{
????????????//?處理完成操作
????????}
????????.collect?{

????????}
}

除此以外，我們還可以通過(guò)捕獲式 try {} finally {} 去獲取完成情況。

4. Flow的特點(diǎn)

我們?cè)趯?duì) Flow 已經(jīng)有了一些基礎(chǔ)的認(rèn)知了，再來(lái)聊一聊 Flow 的特點(diǎn)，F(xiàn)low 具有以下特點(diǎn)：

• 冷流
• 有序
• 協(xié)作取消

如果你對(duì) Kotlin 中的 Sequence 有一些認(rèn)識(shí)，那么你應(yīng)該可以輕松的 Get 到前兩個(gè)點(diǎn)。

冷流

有點(diǎn)類(lèi)似于懶加載，當(dāng)我們觸發(fā) collect 方法的時(shí)候，數(shù)據(jù)才開(kāi)始發(fā)射。

lifecycleScope.launch?{
????val?flow?=?(1..10).asFlow().flowOn(Dispatchers.Main)

????flow.collect?{?num?->
????????????//?具體的消費(fèi)處理
????????????//?...
????????}
????}
}

也就是說(shuō)，在第2行的時(shí)候，雖然流創(chuàng)建好了，但是數(shù)據(jù)一直到第四行發(fā)生 collect 才開(kāi)始發(fā)射。

有序

看代碼比較容易理解：

lifecycleScope.launch?{
????flow?{
????????for(i?in?1..3)?{
????????????Log.e("Flow","$i?emit")
????????????emit(i)
????????}
????}.filter?{
????????Log.e("Flow","$it?filter")
????????it?%?2?!=?0
????}.map?{
????????Log.e("Flow","$it?map")
????????"${it?*?it}?money"
????}.collect?{
????????Log.e("Flow","i?get?$it")
????}
}

得到的日志：

E/Flow:?1?emit
E/Flow:?1?filter
E/Flow:?1?map
E/Flow:?i?get?1?money
E/Flow:?2?emit
E/Flow:?2?filter
E/Flow:?3?emit
E/Flow:?3?filter
E/Flow:?3?map
E/Flow:?i?get?9?money

從日志中，我們很容易得出這樣的結(jié)論，每個(gè)數(shù)據(jù)都是經(jīng)過(guò) emit、filter 、map和 collect 這一套完整的處理流程后，下個(gè)數(shù)據(jù)才會(huì)開(kāi)始處理，而不是所有的數(shù)據(jù)都先統(tǒng)一 emit，完了再統(tǒng)一 filter，接著 map，最后再 collect。

協(xié)作取消

Flow 采用和協(xié)程一樣的協(xié)作取消，也就是說(shuō)，F(xiàn)low 的 collect 只能在可取消的掛起函數(shù)中掛起的時(shí)候取消，否則不能取消。

如果我們想取消 Flow 得借助 withTimeoutOrNull 之類(lèi)的頂層函數(shù)，不妨猜一下，下面的代碼最終會(huì)打印出什么？

lifecycleScope.launch?{
????val?f?=?flow?{
????????for?(i?in?1..3)?{
????????????delay(500)
????????????Log.e(TAG,?"emit?$i")
????????????emit(i)
????????}
????}
????withTimeoutOrNull(1600)?{
????????f.collect?{
????????????delay(500)
????????????Log.e(TAG,?"consume?$it")
????????}
????}
????Log.e(TAG,?"cancel")
}

5. 操作符對(duì)比

限于篇幅，我僅介紹一下 Flow 中操作符的作用，就不一一介紹每個(gè)操作符具體怎么使用了。

普通操作符：

特殊的操作符

總會(huì)有一些特殊的情況，比如我只需要取前幾個(gè)，我只要最新的數(shù)據(jù)等，不過(guò)在這些情況下，數(shù)據(jù)的發(fā)射就是并發(fā)執(zhí)行的。

組合操作符

展平流操作符

展平流有點(diǎn)類(lèi)似于 RxJava 中的 flatmap，將你發(fā)射出去的數(shù)據(jù)源轉(zhuǎn)變?yōu)榱硪环N數(shù)據(jù)源。

末端操作符

顧名思義，就是幫你做 collect 處理，collect 是最基礎(chǔ)的末端操作符。

其他還有一些操作符，我這里就不一一介紹了，感興趣可以查看 API。

三、通道

Channel是一個(gè)面向多協(xié)程之間數(shù)據(jù)傳輸?shù)?BlockQueue。它的使用方式超級(jí)簡(jiǎn)單：

lifecycleScope.launch?{
????//?1.?生成一個(gè)?Channel
????val?channel?=?Channel()

????//?2.?Channel?發(fā)送數(shù)據(jù)
????launch?{
????????for(i?in?1..5){
????????????delay(200)
????????????channel.send(i?*?i)
????????}
????????channel.close()
????}
????????????
????//?3.?Channel?接收數(shù)據(jù)
????launch?{
????????for(?y?in?channel)
????????????Log.e(TAG,?"get?$y")
????}
}

實(shí)現(xiàn)協(xié)程之間的數(shù)據(jù)傳輸需要三步：

1.創(chuàng)建 Channel

創(chuàng)建的 Channel的方式可以分為兩種：

• 直接創(chuàng)建對(duì)象：方式跟上述代碼一致。
• 擴(kuò)展函數(shù) produce

如果使用了擴(kuò)展函數(shù)，代碼就變成了：

lifecycleScope.launch?{
????//?1.?生成一個(gè)?Channel
????val?channel?=?produce?{
????????for(i?in?1..5){
????????????delay(200)
????????????send(i?*?i)
????????}
????????close()
????}

????//?2.?接收數(shù)據(jù)
????//?...?省略?跟之前代碼一致
}

直接將第一步和第二步合并了。

2. 發(fā)送數(shù)據(jù)

發(fā)送數(shù)據(jù)使用的 ?Channel#send() 方法，當(dāng)我們數(shù)據(jù)發(fā)送完畢的時(shí)候，可以使用 Channel#close() 來(lái)表明通道已經(jīng)結(jié)束數(shù)據(jù)的發(fā)送。

3. 接收數(shù)據(jù)

正常情況下，我們僅需要調(diào)用 Channel#receive() 獲取數(shù)據(jù)，但是該方法只能獲取一次傳遞的數(shù)據(jù)，如果我們僅需獲取指定次數(shù)的數(shù)據(jù)，可以這么操作：

repeat(4){
????Log.e(TAG,?"get?${channel.receive()}")
}

但如果發(fā)送的數(shù)據(jù)不可以預(yù)估呢？這個(gè)時(shí)候我們就需要迭代 Channel 了

for(?y?in?channel)
????Log.e(TAG,?"get?$y")

四、多協(xié)程數(shù)據(jù)處理

多協(xié)程處理并發(fā)數(shù)據(jù)的時(shí)候，原子性同樣也得不到保證，協(xié)程中出了一種叫 Mutex 的鎖，區(qū)別是它的 lock 操作是掛起的，非阻塞的，感興趣的同學(xué)可以自行查看。

總結(jié)

個(gè)人感覺(jué)協(xié)層的主要作用是簡(jiǎn)化代碼的邏輯，減少了代碼的回調(diào)地獄，結(jié)合 Kotlin，既可以寫(xiě)出優(yōu)雅的代碼，還能降低我們犯錯(cuò)的概率。至于提升多協(xié)程開(kāi)發(fā)的性能？

如果覺(jué)得本文不錯(cuò)，**「三連」**是對(duì)我最大的鼓勵(lì)。我將會(huì)在下一篇文章中和大家討論協(xié)程的原理，歡迎大家關(guān)注。

學(xué)習(xí)協(xié)程和 kotlin 還是很有必要的，我們團(tuán)隊(duì)在開(kāi)發(fā)新的功能的時(shí)候，也全部選擇了 Kotlin。

關(guān)于我

我是九心，新晉互聯(lián)網(wǎng)碼農(nóng)，如果想要進(jìn)階和了解更多的干貨，歡迎關(guān)注我的公眾號(hào)接收到的我的最新文章。

參考文章：

“

《最全面的Kotlin協(xié)程: Coroutine/Channel/Flow 以及實(shí)際應(yīng)用》
《Kotlin中文站》
《Kotlin 的協(xié)程用力瞥一眼》