Flow 简单使用

通过 Flow，我们可以以响应式的编程方式进行协程代码的编写。Flow 类似于协程版本的 RxJava，但是比起 RxJava，它会更加简单，更加容易上手。

基本使用

        GlobalScope.launch(Dispatchers.Main) {
            flow {
                repeat(3) { // 重复3次
                    Thread.sleep(1000) //模拟耗时计算工作
                    emit(it) // 发送结果
                }
            }
                .flowOn(Dispatchers.Default) // 切换到工作子线程
                .collect {
                    // 运行在调用 collect 的线程当中，在此 case 中是 Dispatchers.Main，也就是主线程。
                    Log.i("TAG", "it: $it") //打印 0 1 2
                }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13

以上面的代码为例，我们使用 flow 主要有以下几步：
1.定义生产者函数，在函数中将结果通过 emit 函数发出。
2.如果需要指定生产者函数所运行的线程，通过 flowOn 函数指定并传入一个 CoroutineContext。
3.调用 collect 函数收集生产者函数的计算结果，collect 是一个 suspend 函数，运行在调用 collect 函数的线程当中，且在收集完成之前会一直挂起，因此需要在挂起函数中被调用。

操作符

flow 提供了非常丰富且便捷的操作符。

flowOn

用于指定生产者函数运行在哪个 CoroutineContext，如第一个例子所示。

onEach

每当生产者函数发送结果时，都会被调用。

    flow {
            repeat(3) {
                emit(it)
        }
    }.onEach {
        println("onEach $it") // 打印 0 1 2
    }.collect {
        println("collect $it") // 打印 0 1 2
    }
1
2
3
4
5
6
7
8
9

launchIn

指定在哪个 CoroutineContext 中进行收集，可以从源码中看出它的实现。

public fun  Flow.launchIn(scope: CoroutineScope): Job = scope.launch {
    collect() // tail-call
}
1
2
3

通过 launchIn，我们可以将第一个例子改成如下代码：

        GlobalScope.launch(Dispatchers.Main) {
            flow {
                repeat(3) { // 重复3次
                    Thread.sleep(1000) //模拟耗时计算工作
                    emit(it) // 发送结果
                }
            }
                .flowOn(Dispatchers.Default) // 切换到工作子线程
                .onEach {
                    // 运行在调用 launchIn 的线程当中，在此 case 中是 Dispatchers.Main，也就是主线程。
                    Log.i("TAG", "it: $it") //打印 0 1 2
                }
                .launchIn(this)
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

onEmpty

当生产者函数没有发送任何结果时会被调用。

    flow { }
        .onEmpty { emit(-1) }
        .collect { println("testOnEmpty: $it") } // 打印 -1
        
    flow { emit(1) }
        .onEmpty { emit(-1) }
        .collect { println("testOnEmpty: $it") } // 打印 1
1
2
3
4
5
6
7

onStart

在收集开始之前调用

    // 打印 start 0 1 2
    flow {
        repeat(3) {
            emit(it)
        }
    }
        .onStart { println("start") }
        .collect { println("$it") }
1
2
3
4
5
6
7
8

onCompletion

在收集完成后调用

    // 打印 0 1 2 completion 
    flow {
        repeat(3) {
            emit(it)
        }
    }
        .onCompletion { println("completion") }
        .collect { println("$it") } 
1
2
3
4
5
6
7
8

catch

当上游操作符或者生产者函数出现异常时被调用：

        // 打印
        // collect: 0
        // catch: java.lang.NullPointerException
        flow {
            repeat(3) {
                if (it == 1) {
                    throw NullPointerException()
                }
                emit(it)
            }
        }
            .catch { println("catch: $it") }
            .collect { println("collect: $it") }
1
2
3
4
5
6
7
8
9
10
11
12
13

map

将上游的结果进行变换后，发送给下游

        // 打印 0 1 4 9 16
        flow {
            repeat(5) {
                emit(it)
            }
        }
            .map { it * it }
            .collect { println("$it") }
1
2
3
4
5
6
7
8

flatMapConcat

根据上游 flow 的值生成新的 flow，再将新的 flow 的结果发送给下游收集者，下面是打印99乘法表的例子

        //打印
        //1 2 3 4 5 6 7 8 9
        //2 4 6 8 10 12 14 16 18
        //3 6 9 12 15 18 21 24 27
        //4 8 12 16 20 24 28 32 36
        //5 10 15 20 25 30 35 40 45
        //6 12 18 24 30 36 42 48 54
        //7 14 21 28 35 42 49 56 63
        //8 16 24 32 40 48 56 64 72
        //9 18 27 36 45 54 63 72 81
        flow {
            for (i in 1..9) {
                emit(i)
            }
        }
            .flatMapConcat {
                flow {
                    for (j in 1..9) {
                        emit(it * j)
                    }
                    println()
                }
            }
            .collect { print("$it ") }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

collectLatest

当收集函数被挂起时，如果生产者函数发送了新结果，则取消当前被挂起的收集函数，并基于新结果重新运行收集函数。换句话说就是确保只接收最新的数据。

        //打印
        //0 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
        //1 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
        //2 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
        //3 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
        //4 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
        //5 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
        //6 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
        //7 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
        //8 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled
        //9
        flow {
            repeat(10) {
                delay(100)
                emit(it)
            }
        }.collectLatest {
            try {
                delay(200)
            } catch (e: CancellationException) {
                println("$it $e")
                throw e
            }
            println("$it") //到这才算收集成功
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

由于 collect 函数每次收集时都会挂起 200 毫秒，而生产函数每 100 毫秒就会发送一次数据，因此 collect 函数每次的挂起都会因为有新结果而被取消，直到最后一个数据发送时才能成功收集。