一、前语

Kotlin Flow是Kotlin中呼应式编程框架的实现，是Kotlin生态中的一个重要组成部分，而说到呼应式编程框架，作为Android开发的咱们势必会联想RxJava，因其丰厚、强壮的功用，峻峭的学习曲线，让人又爱又恨。相较于RxJava，Flow的一个最大特点是其根据Kotlin协程，一个Flow必须运转在协程中，因而在Flow中咱们能够运用协程供给的特性编写愈加简洁、轻量的异步代码。

1、呼应式编程

关于呼应式编程，在我的了解中，它是一种面向异步 数据流的编程范式。其中数据流的概念很好了解，便是一组按时间顺序排列的数据序列；而异步是指数据的接纳是经过注册的callback/observer/collector来完结的，典型的观察者形式。

正因为呼应式编程面向的是异步数据流，而不是函数，所以能够在此之上抽象出一组强壮的东西函数，让咱们运用少数的代码便能实现杂乱的业务逻辑，比方查找框防抖，多个有依赖联系的网络恳求等等。

这儿给大家推荐一篇介绍呼应式编程的优秀文章：

The introduction to Reactive Programming you’ve been missing

借用文章中的一个比方，假设咱们现在想要计算用户在一次“连击”中对应的点击次数，把250ms内大于等于2次以上的点击视为一次“连击”。在传统的编程模型下，咱们势必要界说一些的变量来记载状态以及编写相应的守时逻辑，可是在呼应式编程中，咱们只需求极少数的代码便能完结上述逻辑。

经过下面的示意图来更直观地了解这一进程：

首先经过buffer(clickStream.throttle(250ms))将250ms内的发生的点击事情聚组成一个list，再运用map('get length of list')将list的数据流通换为list长度的数据流，最后过滤掉长度小于2的数据就得到了终究希望的数据流。

2、基本概念

在进一步之前，先介绍下Flow中的一些基本概念。在官方文档中，将Flow关联的各人物划分为三类：

出产者（Producer）：担任数据的出产、发送；
中介（Intermediary）：可选的，能够有若干个，担任对Flow中的数据，甚至是Flow本身的进行变换；
顾客（Consumer）：担任从Flow中接纳数据。

中介其实便是各类中心操作符：map、filter等，根据人物在数据流中的位置，咱们将其上面的部分称作上游，下面的部分称作下流。

举一个比方，假设咱们有如下的数据流，Intermediary[i] 便是Producer->…->Intermediary[i – 1] 的下流、Intermediary[i + 1] ->…->Consumer的上游：

Producer->...->Intermediary[i- 1 ] ->Intermediary[i] ->Intermediary[i+ 1 ] ->...->Consumer

二、Flow的创建和运用

创建一个Flow十分简略，官方供给了一个顶层函数用于创建Flow，如下所示：

public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T>

这一顶层函数又被称作Flow builder，其中有两个关键的接口：Flow和FlowCollector。

public interface Flow<out T> {
    public suspend fun collect(collector: FlowCollector<T>)
}

public fun interface FlowCollector<in T> {
    /**
     * Collects the value emitted by the upstream.
     * This method is not thread-safe and should not be invoked concurrently.
     */
    public suspend fun emit(value: T)
}

Flow描述一个异步数据流，相似于RxJava中的Observable，而FlowCollector担任数据的接纳，相似于RxJava中的Observer。

Flow的界说很简略，只要一个collect办法，接纳一个FlowCollector目标作为参数。当调用collect办法时，会将界说好的FlowCollector目标传递至上游，上游会运用传递过来的FlowCollector目标的emit办法来发送数据。

回到Flow builder，其接纳一个运用suspend润饰，FlowCollector的扩展函数：block，又被称作Producer block，担任数据的出产，并将出产好的数据经过FlowCollector目标的emit办法发送。

public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T>

来看一个简略的比方：

private val myFlow = flow<Int> producer@{
    var value = 0
    while (true) {
        emit(value++)
        delay(1000)
    }
}
GlobalScope.launch {
    myFlow.collect(object : FlowCollector<Int> {
        override suspend fun emit(value: Int) {
            Log.i(TAG, "receive $value")
        }
    })
}

这儿界说了一个每距离1秒发送一次数据的Flow，因为Flow需求运转在协程中，所以经过GlobalScope.launch启动了一个协程，然后在协程中调用Flow目标的collect办法，并传递一个FlowCollector实例，在其emit办法中接纳上游发送过来的数据。

需求留意的是，经过Flow Builder创建的Flow归于冷流，意味着只要当Flow的终端操作函数被调用时，比方collect函数，才会履行出产者代码，也便是示例中的producer代码块，并且每次调用终端操作函数时都会创建一条新的数据流，互相互不影响，在下面的示例中producer代码块会被履行两次，发生两条数据流：

GlobalScope.launch {
    myFlow.collect consumer1@{ value -> Log.i(TAG, "receive $value") }
}
GlobalScope.launch {
    myFlow.collect consumer2@{ value -> Log.i(TAG, "receive $value") }
}

此外，因为FlowCollector的emit办法并不是线程安全的，所以不答应咱们在一个新的协程或许新的Dispatcher中调用，以下代码都是非法的：

private val myFlow1 = flow<Int> {
    GlobalScope.launch { 
        emit(1)
    }
}
private val myFlow2 = flow<Int> {
    withContext(Dispatchers.IO) {
        emit(1)
    }
}

1、操作符

1.1、生命周期

Flow中供给了onStart、onCompletion操作符用于监听数据流的开端、完毕，以onStart为例：

public fun <T> Flow<T>.onStart(
    action: suspend FlowCollector<T>.() -> Unit
): Flow<T>

参数action是一个运用suspend润饰，FlowCollector的扩展函数，其Receiver直接/间接地持有从下流传递过来的FlowCollector目标，答应咱们调用emit办法向数据流的开头添加额外的数据，如下所示：

GlobalScope.launch {
    flowOf(1, 2)
        .onStart {
            emit(0)
        }
        .onCompletion {
            emit(3)
        }
        .collect {
            Log.i(TAG, "receive $it")
        }
}
// receive 0
// receive 1
// receive 2
// receive 3

1.2、线程切换

前面有说到，不答应在新起的协程/线程中调用FlowCollector的emit办法，假如需求显现地切换出产者代码运转的线程，需求借助flowOn操作符：

public fun <T> Flow<T>.flowOn(context: CoroutineContext): Flow<T>
private val myFlow = flow<Int> producer@{
    var value = 0
    while (true) {
        emit(value++)
        delay(1000)
    }
}
GlobalScope.launch(Dispatchers.Main) {
    myFlow.flowOn(Dispatchers.IO)
        .onCompletion completation@{
            Log.i(TAG, "on complete")
        }
        .collect consumer@{
            Log.i(TAG, "receive $it")
        }
}

flowOn操作符只会切换上游数据流的CoroutineContext，不会影响到下流数据流。所以在上述示例代码中，producer代码块会运转在子线程中，completation、consumer代码块会运转在主线程上。

1.3、反常处理

Flow中反常的处理是经过catch操作符来完结的，上游抛出的反常会被捕获，默许会完毕流的履行：

GlobalScope.launch(Dispatchers.Main) {
    flowOf(1, 2, 3)
        .map {
            if (it == 2) {
                throw RuntimeException("error")
            }
            it
        }.flowOn(Dispatchers.IO)
        .catch { e ->
            Log.i(TAG, "catch exception, ${e.message}")
        }
        .collect {
            Log.i(TAG, "receive $it")
        }
}
// receive 1
// catch exception, error

2、callbackFlow

flow构建 api只适用于同步出产的数据，而针对异步回来的数据，官方供给了另一个顶层函数：

public fun <T> callbackFlow(@BuilderInference block: suspend ProducerScope<T>.() -> Unit): Flow<T>
/**
 * Sender's interface to [Channel].
 */
public interface SendChannel<in E> {
    public suspend fun send(element: E)
    public fun trySend(element: E): ChannelResult<Unit>
    public fun close(cause: Throwable? = null): Boolean
    ......
}

block函数接纳一个ProducerScope目标，其实现了SendChannel接口，答应咱们经过send、trySend办法发送数据。callbackFlow的底层根据Channel，概念上相似于Java中的BlockingQueue，不同之处在于它是非阻塞的，假如发送数据时缓冲区已满，会将发送方协程挂起并加入到内部维护的等候队列中，等候缓冲区有空闲时被唤醒。数据的接纳也是相似的，假如当时缓冲区为空，会将接纳方协程挂起并加入到等候队列中，等到有数据时被唤醒。

下面经过一个比方来给大家介绍一下callbackFlow的运用：

private val myCallbackFlow = callbackFlow {
    val callback = object : MyCallback {
        override fun onNext(value: Int) {
            trySend(value) // trySendBlocking(value)
        }
        override fun onCompletion() {
            close()
        }
        override fun onFailure(throwable: Throwable?) {
            cancel()
        }
    }
    api.register(callback)
    awaitClose { api.unregister(callback) }
}
GlobalScope.launch { 
    myCallbackFlow.collect {
        Log.i(TAG, "receive $it") 
    }
}

trySend办法是send办法的非挂起版别，答应咱们在协程外、非挂起函数中调用，可是它不能确保数据的发送，当缓冲区已满时，发送的数据会被丢掉掉。假如咱们需求确保数据的发送，能够调用trySendBlocking办法，当缓冲区已满时，会阻塞当时线程直至成功发送数据或许对应的Channel、协程被封闭、撤销。

需求留意的是，awaitClose办法被强制要求调用，否则会抛出IllegalStateException反常，这是为了防止因协程完毕导致咱们注册的callback泄露。awaitClose办法会将当时协程挂起，并确保在Channel封闭或许协程撤销时履行咱们传递的代码块，然后完结相关资源的释放。还有一点是，传递给awaitClose办法的代码块不能确保和传递给callbackFlow办法的代码块在同一个线程履行，因而相关的注册、反注册办法需求是线程安全的。

参考

The introduction to Reactive Programming you’ve been missing
Kotlin flows on Android
Kotlin Flow 实际运用
Kotlin 异步 | Flow 使用场景及原理
Kotlin 异步 | Flow 限流的使用场景及原理

瞅一眼Kotlin Flow

一、前语

1、呼应式编程

2、基本概念

二、Flow的创建和运用

1、操作符

1.1、生命周期

1.2、线程切换

1.3、反常处理

2、callbackFlow

参考

作者信息

推广

瞅一眼Kotlin Flow

一、前语

1、呼应式编程

2、基本概念

二、Flow的创建和运用

1、操作符

1.1、生命周期

1.2、线程切换

1.3、反常处理

2、callbackFlow

参考

相关文章

带你封装自己的『权限管理』框架

每日记录自己的Android项目（一）——UI界面

2022-11-14 技术周刊第 3 期

Android Room简单使用，结合Flow监听数据变化

作者信息

推广