Kotlin协程之Flow使用 本章前言 kotlin协程之

发表于

banners_twitter.png

本章前言

这篇文章是kotlin协程系列的时候扩展而来，如果对kotlin协程感兴趣的可以通过下面链接进行阅读、

Kotlin协程基础及原理系列

Flow系列

扩展系列

kotlin协程之Flow的使用

本来Flow这章节个人感觉是不太需要讲解，因为主要还是一些协程知识结合响应式流。这些东西我们在使用RxJava和学习协程的过程中已经掌握。但是最近发现还是不少人问关于Flow的一些知识。

本着授人以鱼不如授人以渔的原则，本章节不单单只是讲解如果使用，也会同步讲解一些实现原理。我们将对Flow使用以及实现原理进行同步讲解，篇幅可能有些过长，可以按需跳着看。

感谢催更大军中的每一位，如果不是你们日复一日的催更，可能就没有这篇文章。

**特别鸣谢群友，感谢你们在每一次的吹水摸鱼中不经意的暗示我：
@傻白嫖 @花落随 @AilurusFulgens @阶前听雨 @少年 @本初子午线 @你知道我是谁吗 @贝塞尔曲线 @直线 @篝火 @一本歪经 @MING 下一个昵称～ @null @Jerry J @想**等等365个群友

image.png

异步流

通过对协程的学习我们知道，挂起函数可以异步的返回单个结果值。比如:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
kotlin复制代码fun test(){
    GlobalScope.launch {
        val withStr = withContext(Dispatchers.Default){
            "a"
        }
        val awaitStr = async {
            "b"
        }
        val list = simple()
        Log.d("test","withStr :$withStr")
        Log.d("test","awaitStr :${awaitStr.await()}")
        Log.d("test","list :$list  ")

    }
}
1
2
3
log复制代码D/test: withStr :a
D/test: awaitStr :b
D/test: list :[1, 2, 3]

即使我们在函数中使用List返回一个集合结果，这样也只能认为是返回一个结果，只不过返回的结果类型是List类型。

那么如果我们想在协程中和使用RxJava一样，通过响应式编程方式如何异步返回多个计算好的值呢。可能有人想到使用序列Sequence进行操作。

1
kotlin复制代码public fun <T> sequence(@BuilderInference block: suspend SequenceScope<T>.() -> Unit): Sequence<T> = Sequence { iterator(block) }

使用序列Sequence确实是可以实现，因为sequence本身接接受的也是一个suspend的挂起函数：

1
2
3
4
5
6
7
8
9
10
11
12
kotlin复制代码private fun simple(): Sequence<Int> = sequence {
    for (i in 1..3) {
        Thread.sleep(100)
        yield(i)
    }
}

fun test() {
    simple().forEach { value ->
        Log.d(TAG, "value :${value}")
    }
}
1
2
3
log复制代码D/carman: value :1
D/carman: value :2
D/carman: value :3

但是这里我我们是不可使用delay挂起函数来做延时的，只能使用Thread.sleep。这是因为sequence接收的是一个SequenceScope的扩展函数，而在SequenceScope类上使用了RestrictsSuspension注解。此注解标记的类和接口在用作扩展挂起函数的接收器时受到限制。这些挂起扩展只能调用这个特定接收器上的其他成员或扩展挂起函数，并且不能调用任意的挂起函数。

1
2
3
4
kotlin复制代码@RestrictsSuspension
public abstract class SequenceScope<in T> internal constructor() {
    //....
}

如果没有这限制的话，可能就会出现在使用下一个元素的时候，还会有切换线程的副作用。同理，如果我们想通过指定调度器，来指定序列创建所在的线程，同样是不可以的，甚至都不可能设置协程上下文。

既然序列Sequence有这么多限制，那么就必须创造有个新的东西来实现，这个时候Flow就应运而生。

image.png

Flow与RxJava区别

对于熟悉响应式流（Reactive Streams）或RxJava这样的响应式框架的人来说。Flow的设计也许看起来会非常熟悉，尤其是各种操作符看起来都近乎一样。

Flow的设计灵感也来源于响应式流以及其各种实现。但是 Flow 的主要目标是拥有尽可能简单的设计，以及对kotlin协程更友好的支持。有兴趣可以看看 Reactive Streams and Kotlin Flows 这篇文章了解Flow的故事。

虽然有所不同，但从概念上讲，Flow 依然是响应式流。和RxJava一样，依然有冷热流之分。相比于RxJava的切换线程，Flow也会更加简单。

官方在 kotlinx.coroutines中提供的相关响应式模块（如：kotlinx-coroutines-reactive 用于 Reactive Streams， kotlinx-coroutines-rx2/kotlinx-coroutines-rx3 用于 RxJava2/RxJava3等）。 这些模块可以让Flow与其他实现之间进行转换。

Flow本身是一个接口，在这个接口里面定义了一个挂起函数collect函数，它接收的是一个FlowCollector对象。FlowCollector接口中有一个挂起函数emit。那它们又是如何实现响应式流的呢。

1
2
3
4
5
6
7
8
kotlin复制代码public interface Flow<out T> {
    @InternalCoroutinesApi
    public suspend fun collect(collector: FlowCollector<T>)
}

public interface FlowCollector<in T> {
    public suspend fun emit(value: T)
}

创建冷数据流Flow

老规矩，现在我们Flow来替换之前的使用序列Sequence的实现：

通过flow {...}函数创建

1
2
3
4
5
6
7
8
9
10
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flow {
            for (i in 1..3) {
                delay(100)
                emit(i)
            }
        }.collect { value -> Log.d(TAG, "value :${value}") }
    }
}

注意使用Flow的代码与先前示例的区别。这里使用的是flow {...} 函数创建了一个冷数据流Flow，通过emit来发射数据，然后通过collect函数来收集这些数据。但是因为collect是挂起函数，挂起函数的调用又必须在另一个挂起函数或者协程作用域中。此时就需要我们使用协程来执行。

我们继续来看看它们具体是如何实现的，上源码：

1
kotlin复制代码public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)

虽然我们使用的是flow {...} 函数，但是实际是通过SafeFlow类创建的Flow对象。SafeFlow继承自AbstractFlow。而AbstractFlow同时继承了Flow和CancellableFlow两个接口。这也就意味着我们创建的冷数据流Flow是可以取消的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
kotlin复制代码private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
    override suspend fun collectSafely(collector: FlowCollector<T>) {
        collector.block()
    }
}

@FlowPreview
public abstract class AbstractFlow<T> : Flow<T>, CancellableFlow<T> {

    public final override suspend fun collect(collector: FlowCollector<T>) {
        val safeCollector = SafeCollector(collector, coroutineContext)
        try {
            collectSafely(safeCollector)
        } finally {
            safeCollector.releaseIntercepted()
        }
    }

    public abstract suspend fun collectSafely(collector: FlowCollector<T>)
}

这里可以看到虽然我们调用的是collect函数，但是实际是通过collectSafely函数执行。调用SafeCollector执行collect的block高阶函数参数。只不过是在出现异常的时候它会执行SafeCollector的releaseIntercepted函数。我们继续往下看SafeCollector的实现。

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
kotlin复制代码internal actual class SafeCollector<T> actual constructor(
    @JvmField internal actual val collector: FlowCollector<T>,
    @JvmField internal actual val collectContext: CoroutineContext
) : FlowCollector<T>, ContinuationImpl(NoOpContinuation, EmptyCoroutineContext), CoroutineStackFrame {
    //...
    override val context: CoroutineContext
        get() = completion?.context ?: EmptyCoroutineContext
    
    override fun invokeSuspend(result: Result<Any?>): Any {
        result.onFailure { lastEmissionContext = DownstreamExceptionElement(it) }
        completion?.resumeWith(result as Result<Unit>)
        return COROUTINE_SUSPENDED
    }
    
    public actual override fun releaseIntercepted() {
        super.releaseIntercepted()
    }
    
    override suspend fun emit(value: T) {
        return suspendCoroutineUninterceptedOrReturn sc@{ uCont ->
            try {
                emit(uCont, value)
            } catch (e: Throwable) {
                lastEmissionContext = DownstreamExceptionElement(e)
                throw e
            }
        }
    }
    
    private fun emit(uCont: Continuation<Unit>, value: T): Any? {
        //...
        return emitFun(collector as FlowCollector<Any?>, value, this as Continuation<Unit>)
    }
}

到这里看过协程原理篇的小伙伴应该很熟悉了，这不就协程的执行、调度、恢复过程嘛。这里就不再重复讲解了。如果有需要的可以自己单独去看看。传送门->协程原理1 传送门->协程原理2。

image.png

通过扩展函数asFlow创建

Flow的创建除了使用flow {...} 函数以外，我们还可以使用asFlow进行创建，如下：

1
2
3
4
5
kotlin复制代码fun test() {
    lifecycleScope.launch {
        (1..3).asFlow().collect { value -> Log.d(TAG, "value :${value}") }
    }
}

其实asFlow最终调用的还是flow {...}，asFlow的扩展函数有很多种，我们这里只是举例：

1
2
3
4
5
6
7
8
9
10
11
kotlin复制代码public fun <T> Array<T>.asFlow(): Flow<T> = flow {
    forEach { value ->
        emit(value)
    }
}
//....
public fun IntRange.asFlow(): Flow<Int> = flow {
    forEach { value ->
        emit(value)
    }
}

通过flowOf函数创建

flowOf只支持单个值或者可变值。同样的最终调用的还是flow {...}。

1
2
3
4
5
6
7
8
kotlin复制代码public fun <T> flowOf(vararg elements: T): Flow<T> = flow {
    for (element in elements) {
        emit(element)
    }
}
public fun <T> flowOf(value: T): Flow<T> = flow {
    emit(value)
}

例如：

1
2
3
4
5
6
7
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flowOf(1, 2, 2, 3).collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}

上面提到通过Flow 是可以取消的，但是Flow好像没有提供取消操作，那么我们该如何取消Flow的执行呢。

其实很简单，我们知道Flow的执行是依赖于collect的，而它又必须在协程当中调用，因此取消Flow的主要依赖于collect所在的协程的状态。所以取消Flow只需要取消它所在的协程即可。

1
2
3
4
5
6
7
8
9
10
11
KOTLIN复制代码fun test() {
    val job = lifecycleScope.launch {
        flow {
            for (i in 1..3) {
                delay(100)
                emit(i)
            }
        }.collect { value -> Log.d(TAG, "value :${value}") }
    }
    job.cancel()
}

是不是突然感觉Flow也没有想象中的那么难搞。不过是在协程的基础上进一步封装。重点来了。为了保证flow上下文的一致性，禁止在flow代码块中出现线程调度的情况的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
kotlin复制代码fun test() {
     lifecycleScope.launch {
        flow {
            for (i in 1..3) {
                delay(100)
                if (i ==2 ){
                    withContext(Dispatchers.IO){
                        //骚操作
                        emit(i)
                    }
                }else{
                    emit(i)
                }
            }
        }.collect { value -> Log.d(TAG, "value :${value}") }
    }
}

上面的代码在编译的时候编译期是不会提示你调用错误的，但是在执行的时候会抛出一个java.lang.IllegalStateException: Flow invariant is violated异常。那么在执行的时候如果想切换线程又该怎么办呢

image.png

Flow的线程切换

在使用Flow的时候如果想切换线程，我们就需要使用Flow的扩展函数flowOn。

1
2
3
4
5
6
7
8
kotlin复制代码public fun <T> Flow<T>.flowOn(context: CoroutineContext): Flow<T> {
    checkFlowContext(context)
    return when {
        context == EmptyCoroutineContext -> this
        this is FusibleFlow -> fuse(context = context)
        else -> ChannelFlowOperatorImpl(this, context = context)
    }
}

flowOn将执行此流的上下文更改为指定上下文。该操作符是可组合的。需要注意的是flowOn只影响前面没有自己上下文的操作符。这个要怎么理解能呢。我们先看默认状态flow是都执行在哪些线程上的：

1
2
3
4
5
6
7
8
9
10
11
12
13
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flow {
            for (i in 1..3) {
                Log.d(TAG, "flow :${ currentCoroutineContext()}")
                delay(100)
                emit(i)
            }
        }.collect { value ->
                Log.d(TAG, "collect:${ currentCoroutineContext()} value :${value}")
            }
    }
}

通过前面的学习我们知道，lifecycleScope的launch默认是主线程执行的，那么按照协程的执行原理，我们可以确定上面例子中所有的执行操作都是在主线程上：

1
2
3
4
5
6
kotlin复制代码D/carman: flow :[StandaloneCoroutine{Active}@78b0fe4, Dispatchers.Main.immediate]
D/carman: collect:[StandaloneCoroutine{Active}@78b0fe4, Dispatchers.Main.immediate] value :1
D/carman: flow :[StandaloneCoroutine{Active}@78b0fe4, Dispatchers.Main.immediate]
D/carman: collect:[StandaloneCoroutine{Active}@78b0fe4, Dispatchers.Main.immediate] value :2
D/carman: flow :[StandaloneCoroutine{Active}@78b0fe4, Dispatchers.Main.immediate]
D/carman: collect:[StandaloneCoroutine{Active}@78b0fe4, Dispatchers.Main.immediate] value :3

这个时候我们使用flowOn切换一下线程再看看，会产生有何不一样的变化。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flow {
            for (i in 1..3) {
                Log.d(TAG, "flow :${ currentCoroutineContext()}")
                delay(100)
                emit(i)
            }
        }.flowOn(Dispatchers.IO)
            .collect { value ->
                Log.d(TAG, "collect:${ currentCoroutineContext()} value :${value}")
            }
    }
}
1
2
3
4
5
6
kotlin复制代码D/carman: flow :[ProducerCoroutine{Active}@78b0fe4, Dispatchers.IO]
D/carman: flow :[ProducerCoroutine{Active}@78b0fe4, Dispatchers.IO]
D/carman: collect:[ScopeCoroutine{Active}@1e865fe, Dispatchers.Main.immediate] value :1
D/carman: flow :[ProducerCoroutine{Active}@78b0fe4, Dispatchers.IO]
D/carman: collect:[ScopeCoroutine{Active}@1e865fe, Dispatchers.Main.immediate] value :2
D/carman: collect:[ScopeCoroutine{Active}@1e865fe, Dispatchers.Main.immediate] value :3

可以看到flow代码块中的执行已经切换到另外一个线程执行。但是collect中的代码依然执行在主线程上。那如果我们再增加一个又会是什么结果呢？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flow {
            for (i in 1..3) {
                Log.d(TAG, "flow :${ currentCoroutineContext()}")
                delay(100)
                emit(i)
            }
        }.flowOn(Dispatchers.IO)
            .map {
                Log.d(TAG, "map :${ currentCoroutineContext()}")
                it
            }.flowOn(Dispatchers.Default)
            .collect { value ->
                Log.d(TAG, "collect:${ currentCoroutineContext()} value :${value}")
            }
    }
}
1
2
3
4
5
6
7
8
9
kotlin复制代码D/carman: flow :[ProducerCoroutine{Active}@78b0fe4, Dispatchers.IO]
D/carman: flow :[ProducerCoroutine{Active}@78b0fe4, Dispatchers.IO]
D/carman: map :[ScopeCoroutine{Active}@cc43a14, Dispatchers.Default]
D/carman: collect:[ScopeCoroutine{Active}@8b702bd, Dispatchers.Main.immediate] value :1
D/carman: flow :[ProducerCoroutine{Active}@78b0fe4, Dispatchers.IO]
D/carman: map :[ScopeCoroutine{Active}@cc43a14, Dispatchers.Default]
D/carman: collect:[ScopeCoroutine{Active}@8b702bd, Dispatchers.Main.immediate] value :2
D/carman: map :[ScopeCoroutine{Active}@cc43a14, Dispatchers.Default]
D/carman: collect:[ScopeCoroutine{Active}@8b702bd, Dispatchers.Main.immediate] value :3

这里我们先跳过map操作符，只看我们本次关注的地方。可以看到在flowOn(Dispatchers.IO)前的flow{...}中的代码是执行在IO线程上的，而在调用flowOn(Dispatchers.Default)并没有改变flow{...}的执行线程，只是改变了没有上下文的map执行线程，使map中的代码块执行在Default线程中。而collect中的代码依然执行在主线程上。

如果这里时候我们把flowOn(Dispatchers.IO)去掉，我们就会发现flow{...}和map中的代码块都将执行在Default线程中。

1
2
3
4
5
6
7
8
9
kotlin复制代码D/carman: flow :[ProducerCoroutine{Active}@3656c4d, Dispatchers.Default]
D/carman: map :[ProducerCoroutine{Active}@3656c4d, Dispatchers.Default]
D/carman: flow :[ProducerCoroutine{Active}@3656c4d, Dispatchers.Default]
D/carman: collect:[ScopeCoroutine{Active}@840cc75, Dispatchers.Main.immediate] value :1
D/carman: map :[ProducerCoroutine{Active}@3656c4d, Dispatchers.Default]
D/carman: flow :[ProducerCoroutine{Active}@3656c4d, Dispatchers.Default]
D/carman: collect:[ScopeCoroutine{Active}@840cc75, Dispatchers.Main.immediate] value :2
D/carman: map :[ProducerCoroutine{Active}@3656c4d, Dispatchers.Default]
D/carman: collect:[ScopeCoroutine{Active}@840cc75, Dispatchers.Main.immediate] value :3

通过四次日志的对比，我们可以做一些总结：

  • flowOn可以将执行此流的上下文更改为指定的上下文。
  • flowOn可以进行组合使用。
  • flowOn只影响前面没有自己上下文的操作符。已经有上下文的操作符不受后面flowOn影响。
  • 不管flowOn如何切换线程,collect始终是运行在调用它的协程调度器上。

image.png

Flow的常用操作符

上面提到Flow的操作符map,实际上collect也是一个操作符。只是他们的责任不一样。根据官方的说法，再结合自身使用感觉，笔者把Flow的操作符主要分为五种（非官方）：

  • 过度操作符：又或者叫做流程操作符，用来区分流程执行到某一个阶段。比如：onStart/onEach/onCompletion。过渡操作符应用于上游流，并返回下游流。这些操作符也是冷操作符，就像流一样。这类操作符本身不是挂起函数。它运行的速度很快，返回新的转换流的定义。
  • 异常操作符：用来捕获处理流的异常。比如：catch,onErrorCollect(已废弃，建议用catch)。
  • 转换操作符：主要做一些数据转换操作。比如：transform/map/filter/flatMapConcat等
  • 限制操作符：流触及相应限制的时候会将它的执行取消。比如：drop/take等
  • 末端操作符：是在流上用于启动流收集挂起函数。collect 是最基础的末端操作符，但是还有另外一些更方便使用的末端操作符。例如：toList、toSet、first、single、reduce、fold等等

流程操作符

  • onStart：在上游流启动之前被调用。
  • onEach：在上游流的每个值被下游发出之前调用。
  • onCompletion：在流程完成或取消后调用，并将取消异常或失败作为操作的原因参数传递。

需要注意的是，onStart在SharedFlow(热数据流)一起使用时，并不能保证发生在onStart操作内部或立即发生在onStart操作之后的上游流排放将被收集。这个问题我们在后面文章的热数据流时讲解。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flow {
            Log.d(TAG, "flow")
            emit(1)
        }.onStart {
            Log.d(TAG, "onStart ")
        }.onEach {
            Log.d(TAG, "onEach :${it}")
        }.onCompletion {
            Log.d(TAG, "onCompletion")
        }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
3
4
5
kotlin复制代码D/carman: onStart 
D/carman: flow
D/carman: onEach :1
D/carman: collect :1
D/carman: onCompletion

可以看到整个执行流程依次是onStart->flow{ ...}->onEach->collect->onCompletion。

异常操作符

上面提到了Flow执行的时候可能会出现异常。我们先修改下代码，在onEach中抛出一个异常信息。再看看代码出现异常后会输出怎样的日志信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flow {
            Log.d(TAG, "flow")
            emit(1)
        }.onStart {
            Log.d(TAG, "onStart ")
        }.onEach {
            Log.d(TAG, "onEach :${it}")
            throw NullPointerException("空指针")
        }.onCompletion { cause ->
            Log.d(TAG, "onCompletion catch $cause")
        }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
3
4
5
6
7
8
kotlin复制代码 D/carman: onStart 
 D/carman: flow
 D/carman: onEach 1
 D/carman: onCompletion catch java.lang.NullPointerException: 空指针
 Process: com.example.myapplication, PID: 31145
    java.lang.NullPointerException: 空指针
    ...
    ...

可以看到在onEach中抛出一个异常后，因为异常导致协程退出，所以collect没有执行，但是执行了onCompletion。这又是怎么回事呢。

onCompletion不应该是在collect后执行吗？为什么没有执行collect，反而执行了onCompletion。这个时候我们需要看下源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
kotlin复制代码public fun <T> Flow<T>.onCompletion(
    action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit
): Flow<T> = unsafeFlow {
    try {
        collect(this)
    } catch (e: Throwable) {
        ThrowingCollector(e).invokeSafely(action, e)
        throw e
    }
    val sc = SafeCollector(this, currentCoroutineContext())
    try {
        sc.action(null)
    } finally {
        sc.releaseIntercepted()
    }
}

可以看到在onCompletion中，通过try/catch 块来捕获了collect方法，然后在catch分支里。通过invokeSafely执行了onCompletion中的代码，然后重新抛出异常。既然onCompletion又重新抛出了异常，那我们又该通过什么方式合理的处理这个异常呢？

在协程基础篇文章中，我们提到通过使用try/catch 块来处理异常。那么看下如何使用try/catch 进行捕获异常。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
kotlin复制代码fun test() {
    lifecycleScope.launch {
        try {
            flow {
                Log.d(TAG, "flow")
                emit(1)
                throw NullPointerException("空指针")
            }.onStart {
                Log.d(TAG, "onStart ")
            }.onEach {
                Log.d(TAG, "onEach ")
            }.onCompletion {
                Log.d(TAG, "onCompletion")
            }.collect { value ->
                Log.d(TAG, "collect :${value}")
            }
        } catch (e: Exception) {
            Log.d(TAG, "Exception : $e ")
        }
    }
}

虽然我们同样的可以使用try/catch来处理异常，但是这种写法是不是看上去没有那么优雅。而且出现异常后，无法再继续往下执行。即使我们在flow {...} 构建器内部使用 try/catch，然后再通过emit中发射，这也是不合理的。因为它是违反异常透明性的。

这个时候我们需要使用catch操作符来保留此异常的透明性，并允许封装它的异常处理。catch操作符的代码块可以分析异常并根据捕获到的异常以不同的方式对其做出反应：

  • 可以使用 throw 重新抛出异常。
  • 可以在catch代码块中通过emit将异常转换为新的值发射出去。
  • 可以将异常忽略，或用日志打印，或使用一些其他代码处理它。

现在我们修改一下代码，去掉try/catch块。然后通过catch操作符来捕获异常后，最后通过emit中发射一个新的值出去。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flow {
            Log.d(TAG, "flow")
            emit(1)
            throw NullPointerException("空指针")
        }.onStart {
            Log.d(TAG, "onStart ")
        }.onEach {
            Log.d(TAG, "onEach ")
        }.catch { cause ->
            Log.d(TAG, "catch $cause")
            emit(2)
        }.onCompletion { cause ->
            Log.d(TAG, "onCompletion catch $cause")
        }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
3
4
5
6
kotlin复制代码D/carman: onStart 
D/carman: flow
D/carman: onEach 1
D/carman: catch java.lang.NullPointerException: 空指针
D/carman: collect :2
D/carman: onCompletion catch null

可以看到我们通过catch操作符捕获异常后，collect能够只能收集到上游发射的值。通过我们在catch操作符中通过emit发射的值2也正常被收集。而且我们在onCompletion也不会收集到异常信息。

这个时候我们如果再修改一下代码，在catch操作符后面再加一个map操作符，通过它再抛出一个新的异常又会是什么情况呢。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flow {
            Log.d(TAG, "flow")
            emit(1)
        }.onStart {
            Log.d(TAG, "onStart ")
        }.onEach {
            Log.d(TAG, "onEach $it")
            throw NullPointerException("空指针")
        }.catch { cause ->
            Log.d(TAG, "catch $cause")
            emit(2)
        }.map {
            Log.d(TAG, "map")
            throw NullPointerException("新的异常")
            it
        }.onCompletion { cause ->
            Log.d(TAG, "onCompletion2 catch $cause")
        }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
3
4
5
6
7
8
9
10
kotlin复制代码D/carman: onStart 
D/carman: flow
D/carman: onEach 1
D/carman: catch java.lang.NullPointerException: 空指针
D/carman: map
D/carman: onCompletion2 catch java.lang.NullPointerException: 新的异常
Process: com.example.myapplication, PID: 32168
java.lang.NullPointerException: 新的异常
...
...

程序直接崩溃了。这又是什么情况。这是因为每个操作符只是针对它上游的流，如果下游的流中出现异常，我们需要再次添加一个catch操作符才能正常捕获。

但是如果我们的异常是在collect末端操作符中出现，这个时候我们就只能通过try/catch整个Flow数据流或来处理，或者通过协程上下文中的CoroutineExceptionHandler来处理（这里可以自己动手试试）。

image.png

转换操作符

在流转换操作符中，最通用的一种称为transform。它可以用来模仿简单的转换。还有像map、fliter、zip、Combine、flatMapConcat、flatMapMerge、flatMapLatest等等

transform操作符

transform操作符任意值任意次，其他转换操作符都是基于transform进行扩展的。比如：可以在执行长时间运行的异步请求之前，发射一个字符串并跟踪这个响应。

1
2
3
4
5
6
7
8
9
10
kotlin复制代码fun test() {
    lifecycleScope.launch {
        (1..3).asFlow().transform {
            emit(it)
            emit("transform $it")
        }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
3
4
5
6
kotlin复制代码D/carman: collect :1
D/carman: collect :transform 1
D/carman: collect :2
D/carman: collect :transform 2
D/carman: collect :3
D/carman: collect :transform 3

map操作符

学过RxJava的同学就比较熟悉，我们同通过map操作符进行数据转换操作，包括转换发射出去的数据的类型：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
kotlin复制代码fun test() {
    lifecycleScope.launch {
        flow {
            emit(1)
        }.map {
            Log.d(TAG, "第一次转换")
            it * 5
        }.map {
            Log.d(TAG, "第一次转换后的值 :$it")
            "map $it"
        }.collect { value ->
            Log.d(TAG, "最终转换后的值 :${value}")
        }
    }
}
1
2
3
kotlin复制代码D/carman: 第一次转换
D/carman: 第一次转换后的值 :5
D/carman: 最终转换后的值 :map 5

可以看到我们在第一个map操作符中进行乘运算，第二map操作符中进行类型转换。最终接收到我们经过多次转换处理后的数据。这样做的好处就是，能够保证我们在每一个流的过程中单一职责，一次转换只执行一种操作，而不是把所有过程集中到一起处理完成以后再下发。

map还有同类型操作符mapNotNull，它会过滤掉空值，只发射不为空的值。

1
2
3
4
5
6
7
8
9
10
kotlin复制代码fun test() {
    val flow = flowOf("one", "two", "three",null, "four")
    lifecycleScope.launch {
        flow.mapNotNull {
            it
        }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
3
4
kotlin复制代码D/carman: collect :one
D/carman: collect :two
D/carman: collect :three
D/carman: collect :four

fliter操作符

顾名思义fliter操作符主要是对数据进行一个过滤，返回仅包含与给定匹配的原始流的值的流。

1
2
3
4
5
6
7
8
9
kotlin复制代码fun test() {
    lifecycleScope.launch {
        (1..3).asFlow().filter {
            it < 2
        }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
kotlin复制代码D/carman: collect :1

fliter还有很多同类型操作符，如：filterNot/filterIsInstance/filterNotNull。

filterNot效果恰恰与fliter想法，它取得是与判断条件相反的值。

1
2
3
4
5
6
7
kotlin复制代码fun test() {
    lifecycleScope.launch {
        (1..3).asFlow().filterNot { it < 2 }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
kotlin复制代码D/carman: collect :2
D/carman: collect :3

zip操作符

zip操作符用于组合两个流中的相关值，与RxJava中的zip功能一样：

1
2
3
4
5
6
7
8
9
10
11
kotlin复制代码fun test() {
    val flow1 = (1..3).asFlow()
    val flow2 = flowOf("one", "two", "three")
    lifecycleScope.launch {
        flow2.zip(flow1) { value1, value2 ->
            "$value1 :$value2"
        }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
3
kotlin复制代码D/carman: collect :1 :one
D/carman: collect :2 :two
D/carman: collect :3 :three

限制操作符

take操作符

take操作符返回包含第一个计数元素的流。当发射次数大于等于count的值时，通过抛出异常来取消执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
kotlin复制代码public fun <T> Flow<T>.take(count: Int): Flow<T> {
    require(count > 0) { "Requested element count $count should be positive" }
    return flow {
        var consumed = 0
        try {
            collect { value ->
                if (++consumed < count) {
                    return@collect emit(value)
                } else {
                    return@collect emitAbort(value)
                }
            }
        } catch (e: AbortFlowException) {
            e.checkOwnership(owner = this)
        }
    }
}
private suspend fun <T> FlowCollector<T>.emitAbort(value: T) {
    emit(value)
    throw AbortFlowException(this)
}

我们通过例子来看一下：

1
2
3
4
5
6
7
8
kotlin复制代码fun test() {
    lifecycleScope.launch {
        (1..3).asFlow().take(2)
            .collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
kotlin复制代码D/carman: collect :1
D/carman: collect :2

takeWhile操作符

takeWhile操作符与filter类似，不过它是当遇到条件判断为false的时候，将会中断后续的操作。

1
2
3
4
5
6
7
8
9
10
11
12
kotlin复制代码fun test() {
    lifecycleScope.launch {
       flowOf(1,1,1,2,3,4,4,5,1,2,2,3,3).map {
           delay(100)
           it
       }.takeWhile {
            it  == 1
        }.collect { value ->
            Log.d(TAG, "collect :${value}")
        }
    }
}
1
2
3
kotlin复制代码D/carman: collect :1
D/carman: collect :1
D/carman: collect :1

可以看到虽然我们在设置的之中有四个1，但是因为在第四个1之前遇到了false的判断，所以取消了后续流的执行。

drop操作符

drop操作符与take恰恰相反，它是丢弃掉指定的count数量后执行后续的流。

1
2
3
4
5
6
7
8
kotlin复制代码fun test() {
   lifecycleScope.launch {
       (1..3).asFlow().drop(2)
           .collect { value ->
           Log.d(TAG, "collect :${value}")
       }
   }
}
1
kotlin复制代码D/carman: collect :3

image.png

末端流操作符

collect是最基础的末端操作符，基本上每一个例子当中我们都是使用collect。接下来我们讲解一下其他的末端操作符。

toList操作符

toList操作符是讲我们的流转换成一个List集合

1
2
3
4
5
6
kotlin复制代码fun test() {
    lifecycleScope.launch {
       val list = (1..5).asFlow().toList()
        Log.d(TAG, "toList :${list}")
    }
}
1
kotlin复制代码D/carman: toList :[1, 2, 3, 4, 5]

到这里我们对于Flow的使用以及在什么情况下，对应的使用哪些操作符已经非常清楚。不过我们还需要补充一点。就是我们在执行流的时候，因为每一次发射都上下游都需要时间去处理，这就会导致我们整个flow的处理时间变成长，那我们应该如何缩短这个时间呢。

image.png

Flow的的缓冲

例如：当我们上游的流的发射很慢，每花费100毫秒才产生一个元素而下游的收集器也非常慢，需要花费300毫秒来处理元素。让我们看看从该流收集三个数字要花费多长时间：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
kotlin复制代码fun test() {
    lifecycleScope.launch {
        val time = measureTimeMillis {
            flow {
                for (i in 1..3) {
                    delay(100)
                    emit(i)
                }
            }.collect { value ->
                delay(300)
                Log.d(TAG, "collect :${value}")
            }
        }
        Log.d(TAG, "Collected in $time ms")
    }
}
1
2
3
4
kotlin复制代码D/carman: collect :1
D/carman: collect :2
D/carman: collect :3
D/carman: Collected in 1273 ms

它会整个收集过程大约需要1300多毫秒(个人设备不一样会有偏差)，这是因为这三个数字，他们每个花费400毫秒。这个时候我们就需要通过buffer操作符来压缩转增时间。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
kotlin复制代码fun test() {
    lifecycleScope.launch {
        val time = measureTimeMillis {
            flow {
                for (i in 1..3) {
                    delay(100)
                    emit(i)
                }
            }.buffer().collect { value ->
                delay(300)
                Log.d(TAG, "collect :${value}")
            }
        }
        Log.d(TAG, "Collected in $time ms")
    }
}
1
2
3
4
kotlin复制代码D/carman: collect :1
D/carman: collect :2
D/carman: collect :3
D/carman: Collected in 1039 ms

虽然他们的运行结果是一样的，但是过buffer操作符来执行时候变得更快了。因为buffer高效地创建了处理流，仅仅需要等待第一个数字产生的 100 毫秒以及处理每个数字各需花费的 300 毫秒。这种方式大约花费了 1000 毫秒来运行。

到处为止，Flow的基础篇就结束了。下一章节我们讲对Flow在Android中更高级的用法StateFlow和 SharedFlow进行讲解。

原创不易。如果您喜欢这篇文章，您可以动动小手点赞收藏image.png。

Android技术交流群，有兴趣的可以加入

关联文章
Kotlin协程基础及原理系列

Flow系列

扩展系列

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%