Kotlin協(xié)程之Flow觸發(fā)與消費示例解析

更新時間：2022年09月30日 09:10:55 作者：LeeDuo.

Kotlin協(xié)程中，當需要消費流時，會調用collect方法，觸發(fā)流的消費，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習吧

示例

代碼如下：

launch(Dispatchers.Main) {
    val task = flow {
        emit(2)
        emit(3)
    }.onEach {
        Log.d("liduo", "$it")
    }
    task.collect()
}

一.Flow的觸發(fā)與消費

在Kotlin協(xié)程：Flow基礎原理的分析中，流的觸發(fā)與消費都是同時進行的。每當調用collect方法時，會觸發(fā)流的執(zhí)行，并同時在collect方法中對流發(fā)出的值進行消費。

而在協(xié)程中，其實還提供了分離流的觸發(fā)與消費的操作——onEach方法。通過使用onEach方法，可以將原本在collect方法中的消費過程的移動到onEach方法中。這樣在構建好一個Flow對象后，不會立刻去執(zhí)行onEach方法，只有當調用collect方法時，才會真正的去觸發(fā)流的執(zhí)行。這樣就實現(xiàn)了流的觸發(fā)與消費的分離。

接下來，將對onEach方法進行分析。

1.onEach方法

onEach方法用于預先構建流的消費過程，只有在觸發(fā)流的執(zhí)行后，才會對流進行消費，代碼如下：

public fun <T> Flow<T>.onEach(action: suspend (T) -> Unit): Flow<T> = transform { value ->
    action(value)
    return@transform emit(value)
}

onEach方法是一個Flow接口的擴展方法，返回一個類型為Flow的對象。Flow方法內部通過transform方法實現(xiàn)。

2.transform方法

transform方法是onEach方法的核心實現(xiàn)，代碼如下：

public inline fun <T, R> Flow<T>.transform(
    @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = flow { // 創(chuàng)建Flow對象
    collect { value -> // 觸發(fā)collect
        return@collect transform(value)
    }
}

transform方法也是Flow接口的擴展方法，同樣會返回一個類型為Flow的對象。并且在transform方法內部，首先構建了一個類型為Flow的對象，并且在這個Flow對象的執(zhí)行體內，調用上游的流的collect來觸發(fā)消費過程，并通過調用參數(shù)transform來實現(xiàn)消費。這個collect方法是一個擴展方法，在Kotlin協(xié)程：Flow基礎原理分析過，因此不再贅述。

這就是onEach方法實現(xiàn)觸發(fā)與消費分離的核心，它將對上游的流的消費過程包裹在了一個新的流內，只有當這個新的流或其下游的流被觸發(fā)時，才會觸發(fā)這個新的流自身的執(zhí)行，從而實現(xiàn)對上游的流的消費。

接下來分析一下流的消費過程。

3.collect方法

collect方法用于觸發(fā)流的消費，我們這里調用的collect方法，是一個無參數(shù)的方法，代碼如下：

public suspend fun Flow<*>.collect(): Unit = collect(NopCollector)

這里的無參數(shù)collect方法是Flow接口的擴展方法。在無參數(shù)collect方法中，調用了另一個有參數(shù)的collect方法，這個有參數(shù)的collect方法在Kotlin協(xié)程：Flow基礎原理中提到過，就是Flow接口中定義的方法，并且傳入了NopCollecor對象，代碼如下：

internal object NopCollector : FlowCollector<Any?> {
    override suspend fun emit(value: Any?) {
        // 什么都不做
    }
}

NopCollecor是一個單例類，它實現(xiàn)了FlowCollector接口，但是emit方法為空實現(xiàn)。

因此，這里會調用onEach方法返回的Flow對象的collect方法，這部分在Kotlin協(xié)程：Flow基礎原理進行過分析，最后會觸發(fā)flow方法中的block參數(shù)的執(zhí)行。而這個Flow對象就是transform方法返回的Flow對象。代碼如下：

public inline fun <T, R> Flow<T>.transform(
    @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = flow { // 創(chuàng)建Flow對象
    collect { value -> // 觸發(fā)collect
        return@collect transform(value)
    }
}

通過上面的transform方法可以知道，在觸發(fā)flow方法中的block參數(shù)執(zhí)行后，會調用collect方法。上面提到transform方法是Flow接口的擴展方法，因此這里有會繼續(xù)調用上游Flow對象的collect方法。這個過程與剛才分析的類似，這里調用的上游的Flow對象，就是我們在示例代碼中通過flow方法構建的Flow對象。

此時，會觸發(fā)上游flow方法中block參數(shù)的執(zhí)行，并在執(zhí)行過程中，通過emit方法將值發(fā)送到下游。

接下來，在transform方法中，collect方法的block參數(shù)會被會被回調執(zhí)行，處理上游發(fā)送的值。這里又會繼續(xù)調用transform方法中參數(shù)的執(zhí)行，這部分邏輯在onEach方法中，代碼如下：

public fun <T> Flow<T>.onEach(action: suspend (T) -> Unit): Flow<T> = transform { value ->
    action(value)
    return@transform emit(value)
}

這里會調用參數(shù)action的執(zhí)行，流在這里最終被消費。同時，onEach方法會繼續(xù)調用emit方法，將上游返回的值再原封不動的傳遞到下游，交由下游的流處理。

二.多消費過程的執(zhí)行

首先看下面這段代碼：

launch(Dispatchers.Main) {
    val task = flow {
        emit(2)
        emit(3)
    }.onEach {
        Log.d("liduo1", "$it")
    }.onEach {
        Log.d("liduo2", "$it")
    }
    task.collect()
}

根據上面的分析，兩個onEach方法會按順序依次執(zhí)行，打印出liduo1：2、liduo2：2、liduo1：3、liduo2：3。就是因為onEach方法會將上游的值繼續(xù)向下游發(fā)送。

同樣的，還有下面這段代碼：

launch(Dispatchers.Main) {
    val task = flow {
        emit(2)
        emit(3)
    }.onEach {
        Log.d("liduo1", "$it")
    }
    task.collect {
        Log.d("liduo2", "$it")
    }
}

這段代碼也會打印出liduo1：2、liduo2：2、liduo1：3、liduo2：3。雖然使用了onEach方法，但也可以調用有參數(shù)的collect方法來對上游發(fā)送的數(shù)據進行最終的處理。