Kotlin 协程 (5/6篇) - 响应式编程 Flow

一、概念

• 场景：List 是同步一次性返回多个值，Sequence 是同步（阻塞）分开返回多个值，Suspend 是异步返回单个值，Flow可以异步返回多个值。
• 冷流：只能在创建的时候定义生产数据的代码，无法在外部通过实例更新数据。创建和中间操作只是对上一步的包装并不会执行，所以是响应式编程（也叫声明式），因此可以调用挂起函数自身却不是suspend修饰，只有消费的时候才真正生产和操作数据（按需生成），所以消费都是suspend修饰的在协程中调用。
• 一些名词：构建→上游（前面的流）→当前操作→下游（后面的流）→消费。
• 有序：元素遵循先进先出原则，是一个接一个操作完，而不是统一操作再进行下一步。
• 协作取消：消费的取消只能在可取消的挂起函数挂起的时候取消（即跟随协程取消）。
• 切换线程：默认生产和消费在同一协程上下文中，消费所在的协程决定了生产所在的线程。对于生产可以通过 flowOn 改变上流执行线程而不会影响下流。对于消费可通过 launchIn 改变流执行的线程（不会影响生产中的flowOn）。
• 背压：当数据的消费速度赶不上生产速度。默认情况下生产一个消费一个是同步交替进行的不存在背压，除非使用flowOn切换线程。由于消费操作是挂起函数（协程能轻松切换线程），挂起恢复的特性会阻塞数据生产的速度不会对下流产生背压。
• 防抖：新值的内容发生变化才会消费，相同会被忽略。

二、生产者 producer

用来生产数据，发送值到流中。只能在创建对象的时候定义值的生产方式，无法后期再操作值的生产。

2.1 flow

创建Flow的基本方法，需要手动调用 emit 发射单个值，调用 emitAll 发射另一个Flow中的值。emit 不是线程安全的，不应该并发调用（可考虑channelFlow），不应该在内部调用withContext切换线程（可考虑flowOn操作符）。

val flow = flow {
    emit(5)
    for (i in 1..3) {
        emit(i)
    }
    emitAll(flowOf(1, 2, 3))
}

2.2 flowOf

快速创建固定值集的Flow（类似 listOf() ）。

val flow = flowOf(1, 2, 3)

2.3 asFlow

遍历其它容器（Array、Range、Sequence、Iterable、Iterator）将元素发送到流中。也可以将普通函数或挂起函数返回的值发送到流中。

val flow = listOf(1, 2, 3).asFlow()

2.4 callbackFlow

将回调API改造成Flow。底层使用的sendChannel，默认容量64满了会挂起直到消费出空位，为了避免生产被挂起可以配置为CONFLATED或者UNLIMITED。

fun showWithFlow(): Flow<Int> = callbackFlow {
    //1.创建Flow并发送值（实现Callback接口）
    val callback = object: NetCallback {
        override fun onNextValue(value: Int) {
            try {
                offer(num)
            } catch(t: Throwable) {...}
        }
        override fun onError(ecxeption: Throwable) {
            cancel(CancellationException("API发生错误", exception))
        }
        override fun onCompleted() = close()
    }
    //2.注册回调（传参使用，并对API进行配置操作）
    getData(callback)
    //3.取消协程并注销回调（用来释放API资源）
    awaitClose {...}
}

2.5 emptyFlow

返回一个空的Flow。

val flow = emptyFlow<Int>()

2.6 channelFlow

比 Channel 更冷（触发collect才开始生产）比 Flow 更热（生产消费可以异步执行）。

val flow = channelFlow<Int> { }

三、中间操作 Intermediaries

执行一些操作，不会立即执行，返回的还是Flow。

3.1 状态监控

注意：链式调用中出现多个onStart { action }时，后出现的 action 会先执行，因为后续 onStart 构建的下游流包在了上游 onStart 的外面，并且 action 会在收集上游流数据之前执行。而这个结论却不能沿用到onCompletion { action }，虽然 onCompletion 构建的下游流也包裹在上游 onCompletion 外面，但是 action 总是在收集上游流之后执行。

3.2 过滤

3.3 截取

3.3.1 保留

val flow = (1..5).asFlow()
flow.take(3).collect { print(",$it") }  //打印：1，2，3
flow.takeWhile { it > 3 }.collect { print(",$it") }  //打印：无内容

3.3.2 丢弃

val flow = (1..5).asFlow()
flow.drop(3).collect { print(",$it") }  //打印：4，5
flow.dropWhile { it == 3 }.collect { print(",$it") }  //打印：1，2，3，4，5

3.4 防抖

flow {
    emit(1)
    emit(2)
    delay(600)
    emit(3)
    delay(100)
    emit(4)
    delay(100)
    emit(5)
}.debounce(500)    //只发送 2、5

flow {
    while(true) {
        emit("弹幕")
    }
}.sample(1000)    //每一秒只取一个值
.flowOn(Dispatches.IO)    //由于是死循环需要放在子线程

3.5 去重

3.6 转换

map 和 filter 底层就是使用的 transform。transform 能 emit 任意次数，flow 中每个值都会循环一遍 transform 中的 emit，而 map 只能对值转换。

val flow = (1..3).asFlow()
//map只能转换元素
flow.map { "[$it-a]" }.collect { print(it) }  //打印：[1-a][2-a][3-a]
//mapNotNull不处理null值
flow{
    emit(1)
    emit(2)
    emit(null)
    emit(3)
}.mapNotNull { it?.plus(1) }.collect { print("$it,") } //打印：2,3,4
//flow中每个值都会循环一遍transform中的emit
flow.transform {
    emit("[$it-a]")
    emit("[$it-b]")
}.collect { print(it) } //打印：[1-a][1-b][2-a][2-b][3-a][3-b]
//transform实现filter和map
flow.transform {
    emit(it + 1)
}.transform {
    if (it % 2 == 0) emit(it)
}.collect { print("$it,") }    //打印：2,4
 
flow.transformLatest {
    delay(1000) //此处不延迟的话会每个值都处理而不是只处理最后一个
    emit(it*10)
}.collect { print("$it,") } //打印：20
 
flow.transformWhile {
    emit(it)
    it != 3
}.collect { print("$it,") } //打印：1,2

3.7 线程切换

生产默认执行在消费处所在的协程指定的线程上，生产一个消费一个是同步的。指定线程后，生产和消费是异步进行。

3.8 背压

数据的消费速度赶不上生产速度。默认情况下生产一个消费一个是同步交替进行的不存在背压，emit和collect都是挂起函数会等待对方准备好，最终耗时是每个值生产和消耗的时间总和。buffer使生产和消费并发运行提高效率，设置缓冲区大小（从0开始算的），消费来不及就先把值缓存起来等待着。

• BufferOverflow.SUSPEND：默认模式，缓冲区满了就和默认情况一样挂起（等待消费），消费一个再生产一个。
• BufferOverflow.DROP_OLDEST：缓冲区满了就丢弃还没消费掉的旧值，存入新值。
• BufferOverflow.DROP_LATEST：缓冲区满了就丢弃后来生产的新值，直到有空位了再往里存又生产的值（不是丢弃掉的那些）。

3.9 重试

(1..3).asFlow().onEach{
    if(it == 3) throw Exception("发生异常")
    delay(100)
    println("生产：$it")
}.retry(2){
    it.message = "有异常"
}.catch{ ex ->
    println("捕获异常：${ex.massage}")    //用it试试
}.collect{
    println("消费：$it")
}

3.10 规约（发送每步计算的结果）

(1..3).asFlow()
    .onEach { delay(200) }
    .scan(1) { acc, v -> acc + v }
    .collect { print("$it,") }    //打印：1，2，4，7

3.11 Android生命周期

3.12 操作多个流

3.12.1 合并

val flow1 = flowOf(1, 2, 3, 4, 5)
val flow2 = flowOf('a', 'b', 'c')
merge(flow1, flow2).collect { print(",$it") }   //打印：1,2,3,4,5,a,b,c
merge(flow2, flow1).collect { print(",$it") }   //打印：a,b,c,1,2,3,4,5
flow1.zip(flow2) { a, b -> "[$a$b]" }.collect { print(it) } //打印：[1a][2b][3c]
flow2.zip(flow1) { a, b -> "[$a$b]" }.collect { print(it) } //打印：[a1][b2][c3]
flow1.combine(flow2) { a, b -> "[$a$b]" }.collect { print(it) } //打印：[1a][2b][3c][4c][5c]
flow2.combine(flow1) { a, b -> "[$a$b]" }.collect { print(it) } //打印：[a1][b2][c3][c4][c5]

fun one(): Flow = flow { emit(a) }
fun two(): Flow = flow { emit(b) }
fun three(): Flow = flow { emit(c) }
 
one()
.zip(two()) {a, b ->
    user.name = a
    user.age = b
    user
}.zip(three()) { user, c ->
    user.gender = c
    user
}.collect { user ->
    //...
}

3.12.2 展平多维流

val flow1 = flowOf(1, 2, 3, 4, 5)
val flow2 = flowOf('a', 'b', 'c', 'd', 'e', 'f')
val flow3 = flowOf(flow1, flow2)
flow3.flattenMerge(2).collect { print(",$it") }  //打印：1,2,3,4,5,a,b,c,d,e,f
flow3.flattenConcat().collect { print(",$it") }  //打印：1,2,3,4,5,a,b,c,d,e,f

3.12.3 组合操作多维流

fun getToken(): Flow = Flow { emit(token) }    //获取token
fun getUserInfo(): Flow = Flow { emit(info) }    //获取用户信息
 
getToken()
.flatMapContact { token ->
    getUserInfo(token)
}.flowOn(Dispatches.IO)
.collect { userInfo ->
    println(userInfo)
}

flowOf(300, 200, 100)
.flatMapMerge {
    flow {
        delay(it.toLong())
        emit("a$it")
        emit("b$it")
    }
}.collect { println(it) }    //打印：a100,b100,a200,b200,a300,b300

flow {
    emit(1)
    delay(150)
    emit(2)
    delay(50)
    emit(3)
}.flatMapLatest {
    flow {
        delay(100)
        emit("$it")
    }
}.collect { println(it) } //打印：1,3

四、消费者 Consumer

最终操作，会触发流的执行，返回的是结果。

4.1 收集

flow {
    var count = 0
    while (true) {
        emit(count)
            delay(1000)
            count++
    }
}.collectLatest {
    println("start $it")
    delay(2000)
    println("end $it")
}    //打印：start 0，start 1，start 2，start 3.....

4.2 线程切换

flowOf(1，2，3)
    .launchIn(CoroutineScope(Dispatchers.IO))    //相当于切换了线程
    .join()

4.3 规约（累计到单个值）

val flow = (1..5).asFlow()
println(flow.reduce({ a, b -> a + b })) //打印：15
println(flow.fold(2, { a, b -> a + b }))    //打印：17

4.4 转换到其它容器

4.5 取最值

五、流的取消

5.1 协作取消

Flow在协程中消费，因此协程的取消会一并取消Flow。Flow{ }形式的创建对 emit() 操作附加了 ensureActive() 以检测协程的取消，但出于性能原因其它构建形式和操作不会自行检查以响应，因此需要手动调用 cancellable() 检查。

val flow = flow { repeat(5) { emit(it) } }
withTimeout(2500) { flow.collect { println(it) } }
 
//该构建方式会对发射值附加取消检测
flow {
    repeat(5) { emit(it) }
}.collect {
    if (it == 3) cancel()
}
//其它构建方式需要手动检查
(1..5).asFlow().cancellable().collect {
    println(it)
    if (it == 3) cancel()
}

六、异常处理

在flow构建器或处理值的时候可能发生异常，异常会向下流动关闭每个处理步骤，如果没有调用catch，未捕获的异常会立即取消该flow，在消费时重新抛出。可以在flow内部使用 try-finally 语句，也可以在flow外部使用 try-catch 语句。在Android开发中 catch() 可以用来展示错误信息或者显示默认数据（例如空列表）。

flow {
    emit(1)
    emit(2)
    throw NullPointerException()
}.map {
    it +1
}.catch {
    println("catch：$it")
}.collect {
    println("collect：$it")
}
//打印：
//collect：2
//collect：3
//catch：java.lang.NullPointerException

七、Flow和Suspend区别

八、一些Android示例

fun updateNews() {
    flow{}
        .onStart { showProgressBar() }
        .onCompletion { hideProgressBar() }
        .onEach { view.show(it) }
        .catch { view.handleError(it) }
        .launchIn(viewModelScope)
}

待整理

回调

变换

组合