woman-in-blue-spaghetti-strap-dress-2266519.jpg

一. Kotlin Flow 介紹

Flow 庫是在 Kotlin Coroutines 1.3.2 發(fā)布之后新增的庫。

官方文檔給予了一句話簡單的介紹：

Flow — cold asynchronous stream with flow builder and comprehensive operator set (filter, map, etc);

Flow 從文檔的介紹來看，它有點類似 RxJava 的 Observable。因為 Observable 也有 Cold 、Hot 之分。

二. Flow 基本使用

Flow 能夠返回多個異步計算的值，例如下面的 flow builder :

        flow {
            for (i in 1..5) {
                delay(100)
                emit(i)
            }
        }.collect{
            println(it)
        }

其中 Flow 接口，只有一個 collect 函數(shù)

public interface Flow<out T> {

    @InternalCoroutinesApi
    public suspend fun collect(collector: FlowCollector<T>)
}

如果熟悉 RxJava 的話，則可以理解為 collect() 對應(yīng)subscribe()，而 emit() 對應(yīng)onNext()。

2.1 創(chuàng)建 flow

除了剛剛展示的 flow builder 可以用于創(chuàng)建 flow，還有其他的幾種方式：

flowOf()

    flowOf(1,2,3,4,5)
        .onEach {
            delay(100)
        }
        .collect{
            println(it)
        }

asFlow()

    listOf(1, 2, 3, 4, 5).asFlow()
        .onEach {
            delay(100)
        }.collect {
            println(it)
        }

channelFlow()

    channelFlow {
        for (i in 1..5) {
            delay(100)
            send(i)
        }
    }.collect{
        println(it)
    }

最后的 channelFlow builder 跟 flow builder 是有一定差異的。

flow 是 Cold Stream。在沒有切換線程的情況下，生產(chǎn)者和消費者是同步非阻塞的。
channel 是 Hot Stream。而 channelFlow 實現(xiàn)了生產(chǎn)者和消費者異步非阻塞模型。

下面的代碼，展示了使用 flow builder 的情況，大致花費1秒：

fun main() = runBlocking {

    val time = measureTimeMillis {
        flow {
            for (i in 1..5) {
                delay(100)
                emit(i)
            }
        }.collect{
            delay(100)
            println(it)
        }
    }

    print("cost $time")
}

flow.png

使用 channelFlow builder 的情況，大致花費700毫秒：

fun main() = runBlocking {

    val time = measureTimeMillis{
        channelFlow {
            for (i in 1..5) {
                delay(100)
                send(i)
            }
        }.collect{
            delay(100)
            println(it)
        }
    }

    print("cost $time")
}

channelFlow.png

當(dāng)然，flow 如果切換線程的話，花費的時間也是大致700毫秒，跟使用 channelFlow builder 效果差不多。

fun main() = runBlocking {

    val time = measureTimeMillis{
        flow {
            for (i in 1..5) {
                delay(100)
                emit(i)
            }
        }.flowOn(Dispatchers.IO)
            .collect {
                delay(100)
                println(it)
            }
    }

    print("cost $time")
}

2.2 切換線程

相比于 RxJava 需要使用 observeOn、subscribeOn 來切換線程，flow 會更加簡單。只需使用 flowOn，下面的例子中，展示了 flow builder 和 map 操作符都會受到 flowOn 的影響。

    flow {
        for (i in 1..5) {
            delay(100)
            emit(i)
        }
    }.map {
        it * it
    }.flowOn(Dispatchers.IO)
        .collect {
            println(it)
        }

而 collect() 指定哪個線程，則需要看整個 flow 處于哪個 CoroutineScope 下。

例如，下面的代碼 collect() 則是在 main 線程：

fun main() = runBlocking {

    flow {
        for (i in 1..5) {
            delay(100)
            emit(i)
        }
    }.map {
        it * it
    }.flowOn(Dispatchers.IO)
        .collect {
            println("${Thread.currentThread().name}: $it")
        }
}

執(zhí)行結(jié)果：

main: 1
main: 4
main: 9
main: 16
main: 25

值得注意的地方，不要使用 withContext() 來切換 flow 的線程。

2.3 flow 取消

如果 flow 是在一個掛起函數(shù)內(nèi)被掛起了，那么 flow 是可以被取消的，否則不能取消。

fun main() = runBlocking {

    withTimeoutOrNull(2500) {
        flow {
            for (i in 1..5) {
                delay(1000)
                emit(i)
            }
        }.collect {
            println(it)
        }
    }

    println("Done")
}

執(zhí)行結(jié)果：

1
2
Done

2.4 Terminal flow operators

Flow 的 API 有點類似于 Java Stream 的 API。它也同樣擁有 Intermediate Operations、Terminal Operations。

Flow 的 Terminal 運算符可以是 suspend 函數(shù)，如 collect、single、reduce、toList 等；也可以是 launchIn 運算符，用于在指定 CoroutineScope 內(nèi)使用 flow。

@ExperimentalCoroutinesApi // tentatively stable in 1.3.0
public fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job = scope.launch {
    collect() // tail-call
}

整理一下 Flow 的 Terminal 運算符

• collect
• single/first
• toList/toSet/toCollection
• count
• fold/reduce
• launchIn/produceIn/broadcastIn

該系列的相關(guān)文章：

Kotlin Coroutines Flow 系列(二) Flow VS RxJava2
Kotlin Coroutines Flow 系列(三) 異常處理
Kotlin Coroutines Flow 系列(四) 線程操作
Kotlin Coroutines Flow 系列(五) 其他的操作符