Kotlin Flow（一）基本使用

怼烎@ 2023-07-02 10:25 51阅读 0赞

Kotlin协程中使用挂起函数（Suspend函数）可以异步地返回单个计算结果，但是如果有多个计算结果希望通过协程的方式异步返回，这时可以使用Flows（基于kotlin v. 1.3.61）。

## 返回多个值的方式 ##

### 使用Collections ###

一般我们可以使用集合类存储多个值，例如foo()返回一个list，包含3各成员，可以通过foreach对其遍历并打印结果

fun foo(): List<Int> = listOf(1, 2, 3)
    
    fun main() {
        foo().forEach { value -> println(value) } 
    }

输出：

1
    2
    3

### 使用Sequences ###

除了list以外，我们可以使用squences对单个数值计算后逐条返回，但是计算过程是同步的

fun foo(): Sequence<Int> = sequence { // sequence builder
        for (i in 1..3) {
            Thread.sleep(100) // pretend we are computing it
            yield(i) // yield next value
        }
    }
    
    fun main() {
        foo().forEach { value -> println(value) } 
    }

打印结果相同，每100ms输出一个

### 使用挂起函数 Suspending functions ###

上面的计算会阻塞UI线程，此时我们会考虑使用协程挂起的方式返回结果：

suspend fun foo(): List<Int> {
        delay(1000) // pretend we are doing something asynchronous here
        return listOf(1, 2, 3)
    }
    
    fun main() = runBlocking<Unit> {
        foo().forEach { value -> println(value) } 
    }

1s后结果被一次性输出

### 使用Flow ###

使用 List<Int> 意味着我们的结果只能一次性返回，此时还有一个选择就是使用Flow<Int>，可以像Sequence<Int>一样逐条计算后返回流式结果，同时计算可以在异步完成，不会阻塞UI

fun foo(): Flow<Int> = flow { // flow builder
        for (i in 1..3) {
            delay(100) // pretend we are doing something useful here
            emit(i) // emit next value
        }
    }
    
    fun main() = runBlocking<Unit> {
        // Launch a concurrent coroutine to check if the main thread is blocked
        launch {
            for (k in 1..3) {
                println("I'm not blocked $k")
                delay(100)
            }
        }
        // Collect the flow
        foo().collect { value -> println(value) } 
    }

数值和“I'm not blocked”并行打印，说明collect并没有阻塞UI

I'm not blocked 1
    1
    I'm not blocked 2
    2
    I'm not blocked 3
    3

通过上面例子可以看到，Flow有以下特征:

*  flow\{ ... \} 构建一个Flow类型
 *  flow \{ ... \}内可以使用suspend函数.
 *  foo()不需要是suspend函数
 *  emit方法用来发射数据
 *  collect方法用来遍历结果

## Flow 是冷流 ##

Flows像Sequences一样是冷流，即在调用collect之前，flow\{ ... \}中的代码不会执行

fun foo(): Flow<Int> = flow { 
        println("Flow started")
        for (i in 1..3) {
            delay(100)
            emit(i)
        }
    }
    
    fun main() = runBlocking<Unit> {
        println("Calling foo...")
        val flow = foo()
        println("Calling collect...")
        flow.collect { value -> println(value) } 
        println("Calling collect again...")
        flow.collect { value -> println(value) } 
    }

输出结果:

Calling foo...
    Calling collect...
    Flow started
    1
    2
    3
    Calling collect again...
    Flow started
    1
    2
    3

这也是 foo() 不需要标记为suspend的原因，因为foo()中的执行不需要挂起，可以很快返回结果，等到调用collect的时候才开始执行。

## Flow 的取消 ##

Flow创建后并不返回可以cancel的句柄，但是一个flow的collect是suspend的，所以可以像取消一个suspend方法一样取消flow的collection。就好像RxJava中没有subscribe之前仅仅创建一个Observable是不需要unsubscribe的

fun foo(): Flow<Int> = flow { 
        for (i in 1..3) {
            delay(100)          
            println("Emitting $i")
            emit(i)
        }
    }
    
    fun main() = runBlocking<Unit> {
        withTimeoutOrNull(250) { // Timeout after 250ms 
            foo().collect { value -> println(value) } 
        }
        println("Done")
    }

输出结果：

Emitting 1
    1
    Emitting 2
    2
    Done

## Flow 的创建 ##

从前面的例子中我们知道可以通过 flow \{ ... \} 创建Flow。除此之外还有以下两种方式

*  flowOf 创建一个保护固定数量的flow，类似listOf
 *  任意集合类或者squence通过.asFlow()转成一个flow

例如可以将一个IntRange转成一个flow

(1..3).asFlow().collect { value -> println(value) }