前言

上周在內(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é)程中繞不過(guò)的Flow》

目錄

目錄

一、基礎(chǔ)

1. 概念

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

Essentially, coroutines are light-weight threads.

協(xié)程是輕量級(jí)的線程，為什么是輕量的瞎嬉？可以先告訴大家結(jié)論蝎毡，因?yàn)樗?strong>基于線程池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. 銷毀的時(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)度器類型有：

• 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)有減少同云，只是交給了其他線程：

協(xié)程請(qǐng)求數(shù)據(jù)過(guò)程

launch

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

除了 launch秸滴，還有一個(gè)方法跟它很像武契，就是 async，它的作用是創(chuàng)建一個(gè)協(xié)程荡含，之后返回一個(gè) Deferred<T>對(duì)象咒唆，我們可以調(diào)用 Deferred#await()去獲取返回的值，有點(diǎn)類似于 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. 銷毀的時(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 的使用類似于 RxJava誉券，現(xiàn)在我收回這句話，事實(shí)上刊愚，LiveData 更加簡(jiǎn)單和純粹踊跟，它建立單一的生產(chǎn)消費(fèi)模型，F(xiàn)low 才是類似于 RxJava 的存在鸥诽。

1. 基礎(chǔ)

先上一段代碼：

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

我在 createFlow 這個(gè)方法中商玫，返回了 Flow<Int> 的對(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<Int> = flow {
    for (i in 1..10)
        emit(i)
}

一直調(diào)用 emit 可能不便捷地回，因?yàn)?RxJava 提供了 Observable.just() 這類的操作符，顯然俊鱼，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<T>
    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)類似于懶加載，當(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 之類的頂層函數(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)類似于 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<Int>()

    // 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<Int> {
        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坯门。

參考文章：

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