Kotlin 的 Flow 用于流式编程。
Flow 基本使用
fun main() {
runBlocking {
flow {
for (i in 1..5) {
delay(100)
emit(i)
}
}.collect {
println(it)
}
}
}
输出：
1
2
3
4
5
Flow 生命周期
fun main() {
runBlocking {
flow {
for (i in 1..5) {
delay(100)
emit(i)
}
}.onStart {
println("onStart")
}.onCompletion {
println("onCompletion")
}.collect {
println(it)
}
}
}
输出：
onStart
1
2
3
4
5
onCompletion
Flow 发生异常时的生命周期
fun main() {
runBlocking {
flow {
for (i in 1..5) {
delay(100)
emit(i)
if (i == 3) throw Exception("my error")
}
}.onStart {
println("onStart")
}.onCompletion {
println("onCompletion")
}.collect {
println(it)
}
}
}
输出：
onStart
1
2
3
onCompletion
Exception in thread "main" java.lang.Exception: my error
可以看出，Flow 发生异常时，也会先回调 onCompletion，再抛出异常。
Flow 的 onCompletion 可以携带异常信息
fun main() {
runBlocking {
flow {
for (i in 1..5) {
delay(100)
emit(i)
if (i == 3) throw Exception("my error")
}
}.onStart {
println("onStart")
}.onCompletion { error ->
println("onCompletion $error")
}.collect {
println(it)
}
}
}
输出如下：
onStart
1
2
3
onCompletion java.lang.Exception: my error
Exception in thread "main" java.lang.Exception: my error
at ...
需要注意的是，catch 和 onCompletion 一起使用时，两者的顺序会影响运行结果：
- 如果在 onCompletion 之前调用 catch，只有 catch 中能收到异常信息，onCompletion 中将收不到异常信息
- 如果在 onCompletion 之后调用 catch，onCompletion 和 catch 中都可以收到异常信息
- 单独使用 onCompletion 时，如果出现异常，onCompletion 能捕获到异常。但它并不会 catch 这个异常，而是处理完后把异常抛出去。
Flow 异常处理
fun main() {
runBlocking {
flow {
for (i in 1..5) {
delay(100)
emit(i)
if (i == 3) throw Exception("my error")
}
}.onStart {
println("onStart")
}.onCompletion {
println("onCompletion")
}.catch { exception ->
println("catch:$exception")
}.collect {
println(it)
}
}
}
输出：
onStart
1
2
3
onCompletion
catch:java.lang.Exception: my error
Flow Retry 机制
fun main() {
runBlocking {
flow {
for (i in 1..5) {
delay(100)
emit(i)
if (i == 3) throw Exception("my error")
}
}.retry(3).onStart {
println("onStart")
}.onCompletion {
println("onCompletion")
}.catch { exception ->
println("catch:$exception")
}.collect {
println(it)
}
}
}
输出：
onStart
1
2
3
1
2
3
1
2
3
1
2
3
onCompletion
catch:java.lang.Exception: my error
Flow 超时机制
fun main() {
runBlocking {
withTimeout(200) {
flow {
for (i in 1..5) {
delay(100)
emit(i)
}
}.onStart {
println("onStart")
}.onCompletion {
println("onCompletion")
}.collect {
println(it)
}
}
}
}
输出如下：
onStart
1
onCompletion
Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 200 ms
at ...
Flow 超时处理
fun main() {
runBlocking {
try {
withTimeout(200) {
flow {
for (i in 1..5) {
delay(100)
emit(i)
}
}.onStart {
println("onStart")
}.onCompletion {
println("onCompletion")
}.collect {
println(it)
}
}
} catch (e: TimeoutCancellationException) {
println("Timeout")
}
}
}
输出如下：
onStart
1
onCompletion
Timeout
暂时只能用 try-catch 做超时处理。
Flow 执行时间
fun main() {
runBlocking {
val startTime = System.currentTimeMillis()
flow {
for (i in 1..5) {
delay(100)
println("emit: $i")
emit(i)
}
}.collect {
delay(100)
println("collect: $it")
}
println("Cost: ${System.currentTimeMillis() - startTime}ms")
}
}
输出如下：
emit: 1
collect: 1
emit: 2
collect: 2
emit: 3
collect: 3
emit: 4
collect: 4
emit: 5
collect: 5
Cost: 1105ms
耗时约 1000ms，这是因为 Flow 的生产者和消费者是交替执行的。
Flow 生产者和消费者交替执行
