Kotlin協(xié)程Flow淺析-創(chuàng)新互聯(lián)

Kotlin協(xié)程中的Flow主要用于處理復雜的異步數據，以一種”流“的方式，從上到下依次處理，和RxJava的處理方式類型，但是比后者更加強大。

我們注重客戶提出的每個要求，我們充分考慮每一個細節(jié)，我們積極的做好網站設計、做網站服務，我們努力開拓更好的視野，通過不懈的努力，成都創(chuàng)新互聯(lián)公司贏得了業(yè)內的良好聲譽，這一切，也不斷的激勵著我們更好的服務客戶。 主要業(yè)務：網站建設,網站制作,網站設計,小程序定制開發(fā),網站開發(fā),技術開發(fā)實力，DIV+CSS，PHP及ASP，ASP.Net，SQL數據庫的技術開發(fā)工程師。Flow基本概念

Flow中基本上有三個概念，即 發(fā)送方，處理中間層，接收方，可以類比水利發(fā)電站中的上游，發(fā)電站，下游的概念， 數據從上游開始發(fā)送”流淌“至中間站被”處理“了一下，又流淌到了下游。

示例代碼如下

flow {         // 發(fā)送方、上游
    emit(1)    // 掛起函數，發(fā)送數據
    emit(2)
    emit(3)
    emit(4)
    emit(5)
}
.filter { it >2 }  // 中轉站，處理數據
.map { it * 2 }
.take(2)
.collect{           // 接收方，下游
    println(it)
}
輸出內容：
6
8

通過上面代碼我們可以看到，基于一種鏈式調用api的方式，流式的進行處理數據還是很棒的，接下來具體看一下上面的組成:

• flow{},是個高階函數，主要用于創(chuàng)建一個新的Flow。在其Lambda函數內部使用了emit()掛起函數進行發(fā)送數據。
• filter{}、map{}、take{},屬于中間處理層，也是中間數據處理的操作符，Flow大的優(yōu)勢，就是它的操作符跟集合操作符高度一致。只要會用List、Sequence，那么就可以快速上手 Flow 的操作符。
• collect{},下游接收方，也成為終止操作符，它的作用其實只有一個：終止Flow數據流，并且接收這些數據。

其他創(chuàng)建Flow的方式還是flowOf()函數，示例代碼如下

fun main() = runBlocking{aassssssssaaaaaaaas
    flowOf(1,2,3,4,5).filter { it >2 }
        .map { it * 2 }
        .take(2)
        .collect{
            println("flowof: $it")
    }
}

我們在看一下list集合的操作示例

listOf(1,2,3,4,5).filter { it >2 }
        .map { it * 2 }
        .take(2)
        .forEach{
            println("listof: $it")
        }

通過以上對比發(fā)現，兩者的基本操作幾乎一致，Kotlin也提供了兩者相互轉換的API，Flow.toList()、List.asFlow()這兩個擴展函數，讓數據在 List、Flow 之間來回轉換，示例代碼如下：

//flow 轉list
    flowOf(1,2,3)
        .toList()
        .filter { it >1 }
        .map { it * 2 }
        .take(2)
        .forEach{
            println(it)
        }
    // list 轉 flow
    listOf(1,2,3).asFlow()
        .filter { it >2 }
        .map { it * 2 }
        .take(2)
        .collect{
            println(it)
        }

Flow生命周期

雖然從上面操作看和集合類型，但是Flow還是有些特殊操作符的，畢竟它是協(xié)程的一部分，和Channel不同，Flow是有生命周期的，只是以操作符的形式回調而已，比如onStart、onCompletion這兩個中間操作符。

flowOf(1,2,3,4,5,6)
        .filter {
            println("filter: $it")
            it >3
        }
        .map {
            println("map: $it")
            it * 2
        }
        .take(2)
        .onStart { println("onStart") }
        .collect{
            println("collect: $it")
        }
輸出內容：
onStart
filter: 1
filter: 2
filter: 3
filter: 4
map: 4
collect: 8
filter: 5
map: 5
collect: 10

我們可以看到onStart，它的作用是注冊一個監(jiān)聽事件：當 flow 啟動以后，它就會被回調。

和filter、map、take這些中間操作符不同，他們的順序會影響數據的處理結果，這也很好理解；onStart和位置沒有關系，它本質上是一個回調，不是一個數據處理的中間站。同樣的還有數據處理完成的回調onCompletion。

flowOf(1,2,3,4,5,6)
        .filter {
            println("filter: $it")
            it >3
        }
        .map {
            println("map: $it")
            it * 2
        }
        .take(2)
        .onStart { println("onStart") }
        .onCompletion { println("onCompletion") }
        .collect{
            println("collect: $it")
        }

Flow中onCompletion{} 在面對以下三種情況時都會進行回調：

• 1，Flow 正常執(zhí)行完畢
• 2，Flow 當中出現異常
• 3，Flow 被取消。

處理異常

在數據流的處理過程中，很難保證不出現問題，那么出現異常之后再該怎么處理呢？

• 對于發(fā)生在上游、中間操作這兩個階段的異常，我們可以直接使用 catch 這個操作符來進行捕獲和進一步處理。
• 對于發(fā)生在下游，使用try-catch，把collect{}當中可能出現問題的代碼包裹起來進行捕獲處理。

上游或者中間異常使用catch

fun main() = runBlocking{
    val flow = flow {
        emit(1)
        emit(2)
        throw IllegalStateException()
        emit(3)
    }

    flow.map { it * 2 }
        .catch { println("catch: $it") }
        .collect{
            println("collect: $it")
        }
}
輸出：
collect: 2
collect: 4
catch: java.lang.IllegalStateException

catch 這個操作符的作用是和它的位置強相關的,catch 的作用域，僅限于catch的上游。換句話說，發(fā)生在 catch 上游的異常，才會被捕獲，發(fā)生在 catch 下游的異常，則不會被捕獲。

val flow = flow {
        emit(1)
        emit(2)
        throw IllegalStateException()
        emit(3)
    }

    flow.map { it * 2 }
        .catch { println("catch: $it") }
        .filter { it / 0 >1 } // catch之后發(fā)生異常
        .collect{
            println("collect: $it")
    }
輸出內容:
Exception in thread "main" java.lang.ArithmeticException: / by zero

下游使用try-catch

flowOf(1,2,3)
        .onCompletion { println("onCompletion $it") }
        .collect{
            try {
                println("collect: $it")
                throw IllegalStateException();
            }catch (e: Exception){
                println("catch $e")
            }
        }
輸出：
collect: 1
catch java.lang.IllegalStateException
collect: 2
catch java.lang.IllegalStateException
collect: 3
catch java.lang.IllegalStateException
onCompletion null

切換執(zhí)行線程

Flow適合處理復雜的異步任務，大多數情況下耗時任務放在子線程或線程池中處理，對于UI任務放在主線程中進行。

在Flow中可以使用flowOn操作符實現上述場景中的線程切換。

flowOf(1,2,3,4,5)
        .filter {
            logX("filter: $it")
            it >2 }
        .flowOn(Dispatchers.IO) // 切換線程
        .collect{
            logX("collect: $it")
        }
輸出內容：
================================
filter: 1
Thread:DefaultDispatcher-worker-1
================================
================================
filter: 2
Thread:DefaultDispatcher-worker-1
================================
================================
filter: 3
Thread:DefaultDispatcher-worker-1
================================
================================
filter: 4
Thread:DefaultDispatcher-worker-1
================================
================================
filter: 5
Thread:DefaultDispatcher-worker-1
================================
================================
collect: 3
Thread:main
================================
================================
collect: 4
Thread:main
================================
================================
collect: 5
Thread:main
================================

flowOn 操作符也是和它的位置強相關的。作用域限于它的上游。在上面的代碼中，flowOn 的上游，就是 flowOf{}、filter{} 當中的代碼，所以，它們的代碼全都運行在 DefaultDispatcher 這個線程池當中。只有collect{}當中的代碼是運行在 main 線程當中的。

終止操作符

Flow 里面，最常見的終止操作符就是collect。除此之外，還有一些從集合中借鑒過來的操作符，也是Flow的終止操作符。比如 first()、single()、fold{}、reduce{}，本質上來說說當我們嘗試將 Flow 轉換成集合的時候，已經不屬于Flow的API，也不屬于協(xié)程的范疇了，它本身也就意味著 Flow 數據流的終止。

"冷的數據流"從何而來

在上面文章《Kotlin協(xié)程Channel淺析》中，我們認識到Channel是”熱數據流“，隨時準備好，隨用隨取，就像海底撈里的服務員。

現在我們看下Flow和Channel的區(qū)別

val flow = flow {
        (1..4).forEach{
            println("Flow發(fā)送前：$it")
            emit(it)
            println("Flow發(fā)送后： $it")
        }
    }

    val channel: ReceiveChannel= produce {
        (1..4).forEach{
            println("Channel發(fā)送前： $it")
            send(it)
            println("Channel發(fā)送后： $it")
        }
    }
    
輸出內容：
Channel發(fā)送前： 1

Flow中的邏輯并未執(zhí)行，因此我們可以這樣類比，Channel之所以被認為是“熱”的原因，是因為不管有沒有接收方，發(fā)送方都會工作。那么對應的，Flow被認為是“冷”的原因，就是因為只有調用終止操作符之后，Flow才會開始工作。

除此之外，Flow一次處理一條數據，是個”懶家伙“。

val flow = flow {
        (3..6).forEach {
            println("Flow發(fā)送前：$it")
            emit(it)
            println("Flow發(fā)送后： $it")
        }
    }.filter {
        println("filter: $it")
        it >3
    }.map {
        println("map: $it")
        it * 2
    }.collect {
        println("結果collect: $it")
    }
輸出內容：
Flow發(fā)送前：3
filter: 3
Flow發(fā)送后： 3
Flow發(fā)送前：4
filter: 4
map: 4
結果collect: 8
Flow發(fā)送后： 4
Flow發(fā)送前：5
filter: 5
map: 5
結果collect: 10
Flow發(fā)送后： 5
Flow發(fā)送前：6
filter: 6
map: 6
結果collect: 12
Flow發(fā)送后： 6

相比于滿面春風，熱情服務的Channel,Flow更像個冷漠的家伙，你不找他，他不搭理你。

• Channel,響應速度快，但數據可能是舊的，占用資源
• Flow,響應速度慢，但數據是最新的，節(jié)省資源

Flow也可以是”熱“的，你知道嗎？

你是否還在尋找穩(wěn)定的海外服務器提供商？創(chuàng)新互聯(lián)www.cdcxhl.cn海外機房具備T級流量清洗系統(tǒng)配攻擊溯源，準確流量調度確保服務器高可用性，企業(yè)級服務器適合批量采購，新人活動首月15元起，快前往官網查看詳情吧

分享標題：Kotlin協(xié)程Flow淺析-創(chuàng)新互聯(lián)
文章鏈接：http://muchs.cn/article34/dpsdpe.html

成都網站建設公司_創(chuàng)新互聯(lián)，為您提供企業(yè)網站制作、營銷型網站建設、網站內鏈、定制網站、做網站、App開發(fā)

廣告

聲明：本網站發(fā)布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)