它们的函数注释一言难尽，建议别看，onCompletion的注释不太对。

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
fun main() = runBlocking<Unit> {
    (1..3).asFlow()
        .catch { cause: Throwable ->
            println("0 catch ${cause}")
        }
        .onCompletion { cause: Throwable? ->
            if (cause == null) {
                println("0 Done successfully")
            } else {
                println("0 Done fail")
            }

        }
        .onEach { value ->
            check(value <= 1) { "Crash on $value" }
            println("Got $value")
        }
        .onCompletion { cause: Throwable? ->
            if (cause == null) {
                println("1 Done successfully")
            } else {
                println("1 Done fail")
            }
        }
        .catch { cause: Throwable ->
            println("1 catch ${cause}")
        }
        .onCompletion { cause: Throwable? ->
            if (cause == null) {
                println("2 Done successfully")
            } else {
                println("2 Done fail")
            }
        }
        .collect()// 声明式写法，collect代码块内容写在onEach中。
}

输出结果

Got 1
0 Done fail
1 Done fail
1 catch java.lang.IllegalStateException: Crash on 2
2 Done successfully

catch

  1. 只能捕捉到上游的异常，下游无法（所以它根本不能捕获终端操作符中抛出的异常）。
  2. 只有捕捉到异常，才进入代码块。
  3. 可以抛出其他异常
  4. 可以emit值

onCompletion

这个有点复杂，分情况讨论。

正常结束

  1. onCompletion操作符按声明顺序，依次进入其代码块执行，异常参数为null
  2. 可以emit值

异常结束

异常结束的时候，一个onCompletion操作符异常参数依然有可能为null。

  1. 为null的情况是：该操作符和异常抛出处之间（注意之间），有catch捕获了该异常，且catch没有抛出异常，且终端操作符没有抛出异常。
  2. 其他情况，异常参数都不为null。

当异常参数不为null的情况下（不管它正常结束还是异常结束）才可以在代码块中emit值。

举个例子

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
26
27
28
29
30
31
32
33
fun main() = runBlocking<Unit> {
    (1..3).asFlow()
        .onCompletion { cause: Throwable? ->
            if (cause == null) {
                println("0 Done successfully")
            } else {
                println("0 Done fail")
            }
        }// 0
        .onEach { value ->
            check(value <= 1) { "Crash on $value" }
            println("Got $value")
        }
        .onCompletion { cause: Throwable? ->
            if (cause == null) {
                println("1 Done successfully")
            } else {
                println("1 Done fail")
            }
            emit(100)
        }// 1
        .catch { cause: Throwable ->
            println("1 catch ${cause}")
        }
        .onCompletion { cause: Throwable? ->
            if (cause == null) {
                println("2 Done successfully")
            } else {
                println("2 Done fail")
            }
        }// 3
        .collect{}// 声明式写法，collect代码块内容写在onEach中。
}

上面的示例已异常结束，0/1处异常参数不为空，2处为空（catch把它捕获了）。

如果终端有异常，所有onCompletion操作符异常参数都不为空。

另外，catch和onCompletion的执行顺序按照声明顺序（如果某catch可以执行的话）。

声明式写法

1
2
3
4
5
6
7
8
9
10
11
12
// 1
fun main() = runBlocking<Unit> {
    (1..3).asFlow()
    .onEach { value ->
        check(value <= 1) { "Crash on $value" }
        println("Got $value")
    }.onCompletion {
        println("Done")
    }.catch { e ->
        println("Caught $e")
    }.collect()
}
  1. 为了让catch可以捕获所有异常，把catch放到collect之前，collect留空，业务代码写在onEach中。
  2. onCompletion一定可以执行，上面的写法中，如果上游有异常，异常参数是会不为null，但是我们不要管它，留给catch来处理。这里的执行顺序是onCompletion再catch。
1
2
3
4
5
6
7
8
9
10
11
12
// 2
fun main() = runBlocking<Unit> {
    (1..3).asFlow()
        .onEach { value ->
            check(value <= 1) { "Crash on $value" }
            println("Got $value")
        }.catch { e ->
            println("Caught $e")
        }.onCompletion {
            println("Done")
        }.collect()
}

调换位置。

第一个写法是catch可以捕获到onCompletion的异常，第二种不行。第二种先catch再onCompletion。

抽出一个拓展函数和用法如下：

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
fun main() {
    CoroutineScope(Dispatchers.Default).launch {
        flow<Int> {
            emit(1)
            emit(2)
            emit(3)
        }.catchCompletionCollect(
            action = {
                println("Got $it")
                if (it == 2) {
                    println("throw error")
                    throw IllegalStateException()
                }
            },
            doWhenFlowCompleteSuccessful = {
                println("做一些完全成功才做的事")
            },
            doWhenFlowMeetError = {
                println("meet error: $it")
                println("做一些发生错误才做的事")
            }
        )
    }
    Thread.sleep(99999)
}

suspend fun <T> Flow<T>.catchCompletionCollect(
    action: (T) -> Unit,
    doWhenFlowCompleteSuccessful: () -> Unit,
    doWhenFlowMeetError: (Throwable) -> Unit
) {
    this.onEach {
            action(it)
        }
        .onCompletion { cause: Throwable? ->
            if (cause == null) {
                doWhenFlowCompleteSuccessful()
            }
        }
        .catch {
            doWhenFlowMeetError(it)
        }
        .collect()
}

参考资料：medium