对比Java学Kotlin协程-异步流

Posted 陈蒙_

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了对比Java学Kotlin协程-异步流相关的知识，希望对你有一定的参考价值。

文章目录

异步流定义

我们知道有 RxJava、RxSwift、RxJs，那有 RxKotlin 吗？答案是没有。因为不需要，Kotlin 内置了 「RxKotlin」，天生拥有响应式编程能力，这就是 Kotlin 异步流。
类比 RxJava = 响应式编程+线程，有 Kotlin 异步流 = RxKotlin + 协程。
RxJava 支持切换线程，而Kotlin异步流切换的是协程。
既然是“异步”流，那有“同步”流吗？有的，Kotlin 中的同步流就是 Sequence。
跟 RxJava、Project Reactor 一样，Kotlin Flow 也是响应式编程规范（Reactive Streams Specification）的一种实现。

通过 launch()、async() 这些创建协程的方法可以获取单个异步执行的实例。那如果我们想同时获取多个呢？或者说获取一组异步执行的实例呢？这个时候异步流就派上用场了。

响应式编程与异步流

什么是响应式编程？
抛开具体编程语言，我们看一个赋值语句，涉及 a、b、c 三个变量：

var a = 1
var b = a + 1
a = 2
print(a, b)

试问，a 的值发生改变之后，b 的值会跟着变吗？根据我们的经验，一般 b 是不会变的。其实我们可以设计一门编程语言，让 b 跟着 a 变，这时这个编程语言就是天然支持响应式编程的。
从这个例子我们可以看出，响应式编程的精髓是「响应」，即一方随时「响应」另一方，二者之间可能是依赖关系。

这种响应关系就像水流（flow 或 stream）的上游和下游一样，上游发生变化，下游跟着变化。

操作符

异步流的操作符可以按不同的标准进行分类，比如按出现位置可以分为开头型、中间型和结尾型，比如构建异步流的操作符一般出现在流的开头，而启动型的操作符出现在流的结尾，其他的如转换型的操作符出现在流的中间位置。

构建型

Kotlin 提供了 flow()、flowOf()、asFlow() 构造异步流。后面两个方法本质上还是借助 flow() 完成的。Koltin 异步流中最基本的方法是 flow()、collect() 和 emit()。其中 emit() 方法类似 RxJava 种的 onNext()，每个操作符处理完上游传递给自己的数据然后发射/emit出去，就完成了向下游的传递。

  • flow()，接收一个 FlowCollector emit() 类型的 lambda 表达式，返回一个 Flow collect() 的 lambda 表达式；
  • collect()，相当于下游水龙头，调用 collect() 时触发上游开始执行；
  • emit()，上游产生数据传递给下游；

使用这几个方法，我们就可构造简单的异步流了，举个例子：

fun main() = runBlocking 
    launch 
        for (i in 1..4) 
            println("i am not blocked $i")
            delay(100)
        
    

    simpleFlow().collect  value ->
        println(value)
    


fun simpleFlow(): Flow<Int> = flow 
    for (i in 1..4) 
        delay(1000)
        emit(i)

上述代码输出结果为：

i am not blocked 1
i am not blocked 2
i am not blocked 3
i am not blocked 4
1
2
3
4

需要注意以下几点：

  • simpleFlow() 方法没有 suspend 修饰，因为该方法本身并非一个耗时方法，而是直接返回的同步方法；
  • 通过 emit() 方法发射数据到下游，作用跟普通方法中的 return 和 Sequence 中的 yield() 类似；
  • 只有在调用 collect() 方法时数据流才会启动运行，而且每次调用都会从头执行一遍，否则其前面的其他操作符方法不会执行；
  • collect() 方法是 suspend 的，必须在另一个 suspend 方法中被调用；
  • 数据流可以看做是一组值，每个值的执行也就是流经操作符的顺序是串行的；

关于最后一点，我们看个例子：

fun main() = runBlocking 
    (1..5).asFlow().filter 
        println("filtering $it")
        it % 2 == 0
    .map 
        println("mapping $it")
        "string $it"
    .collect 
        println("collected $it")

结果输出为：

filtering 1
filtering 2
mapping 2
collected string 2
filtering 3
filtering 4
mapping 4
collected string 4
filtering 5

从结果可以看出，每个值都依次经过了 filter()、map() 和 collect()。奇数经过 filter() 时返回 false，就会被被拦下从而没有触发 emit() 方法往下一个操作符发射数据，filter() 和 map() 方法本质上是 inline 方法，最后都会被转换成 transform() 方法，当 filter 条件命中返回 true 时才会调用 emit() 方法，map() 则是无脑调用 emit()。
emit() 方法可以被调用多次：

fun main() = runBlocking 
    (1..4).asFlow()
        .transform  request ->
            emit("making req $request")
            emit(performReq(request))
        
        .collect  response -> println(response) 


suspend fun performReq(req: Int): String 
    delay(1000)
    return "response $req"

上述代码输出：

making req 1
response 1
making req 2
response 2
making req 3
response 3
making req 4
response 4

上下文型

flowOn()

异步流的上下文是指，异步流执行在哪个协程。我们可以粗略的将异步流分为上游和下游两部分，那么上下文要关注的是上游和下游分别运行在哪个协程中。
最常见的场景是，上游是数据生产者，是耗时操作，比如网络请求等，而下游是消费者，比如在主线程中更新UI等。
下游代码，即 collect() 对应的代码，如果没有特殊指定，默认运行在调用该方法的协程中。
我们可以使用 flowOn() 来改变上游代码运行所在的协程，一般是在构造协程的时候。

fun /*context_*/main() = runBlocking 
    simple().flowOn(Dispatchers.Default).collect  value -> log("Collected $value") 


fun simple() = flow 
    for (i in 1..3) 
        delay(100)
        log("Emitting $i")
        emit(i)
    


fun log(msg: String) = println("[$Thread.currentThread().name] $msg")

协作型

Flow的上游好比是泵，下游好比是水龙头。上游抽水的速度和水龙头放水的速度可能不一致，如果想让二者协同合作以达到最高效率或实现某种效果，就需要使用协作型操作符了。

buffer
一般情况下，只有下游完成处理后才会告诉上游继续发射下一个数据。但是如果上游的发射速度高于下游，是否可以不让上游等待呢？这时我们可以使用 buffer()，让上游实现并发。
可以简单的认为，有了buffer之后上游就放飞自我了，发射时机不再受下游限制了，反正提前发射的值可以被缓存（buffer）。
比如上游发射一个数据需要100ms，下游处理一个数据需要200ms：

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.cancellable
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking
import kotlin.time.ExperimentalTime
import kotlin.time.measureTime

@OptIn(ExperimentalTime::class)
fun main(): Unit = runBlocking 

    val elapse = measureTime 
        simpleFlow()/*.buffer()*/.collect 
            println("consuming $it")
            delay(200)
            println("consuming done")
        

        println("all done")
    

    println("elapse: $elapse")


fun simpleFlow() = flow 
    for (i in 1..3) 
        println("producing $i")
        delay(100)
        emit(i)
        println("producing done")

使用buffer前后的耗时情况如下图：

所以未使用buffer时耗时900ms左右，使用buffer之后耗时在700ms左右。

conflate
在上游发射速度快于下游处理速度时，conflate() 可以让我们丢弃中间值，而仅仅处理最新的值。被丢弃的前提是下游还没来得及处理，如果已经开始处理了，则不会被丢弃。当下游的处理耗时是300ms时中间值会被丢弃，而150ms时则会被正常处理：

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlin.time.ExperimentalTime
import kotlin.time.measureTime

@OptIn(ExperimentalTime::class)
fun main(): Unit = runBlocking 

    var delay = 150L // 300L
    
    val elapse = measureTime 

        simpleFlow()/*.buffer()*/.conflate().collect 
            println("consuming $it")
            delay(delay)
            println("consuming $it done")
        

        println("all done")
    

    println("elapse: $elapse")


fun simpleFlow() = flow 
    for (i in 1..3) 
        println("producing $i")
        delay(100)
        emit(i)
        println("producing $i done")

150 的输出：

producing 1
producing 1 done
producing 2
consuming 1
producing 2 done
producing 3
consuming 1 done
consuming 2
producing 3 done
consuming 2 done
consuming 3
consuming 3 done
all done
elapse: 616.934055ms

Process finished with exit code 0

300的输出：

producing 1
producing 1 done
producing 2
consuming 1
producing 2 done
producing 3
producing 3 done
consuming 1 done
consuming 3
consuming 3 done
all done
elapse: 767.395294ms

如果上游和下游速度都慢，我们可以使用 conflate 来提升处理速度，本质上是丢弃中间值。另一种方式是每当上游发射一个值时，下游都取消当前正在处理的任务，重新启动一个新任务处理最新值，这类操作符统称为 xxLatest。

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlin.time.ExperimentalTime
import kotlin.time.measureTime

@OptIn(ExperimentalTime::class)
fun main(): Unit = runBlocking 

    var delay = 150L

    val elapse = measureTime 

        simpleFlow()/*.buffer().conflate()*/.collectLatest 
            println("consuming $it")
            delay(delay)
            println("consuming $it done")
        

        println("all done")
    

    println("elapse: $elapse")


fun simpleFlow() = flow 
    for (i in 1..3) 
        println("producing $i")
        delay(100)
        emit(i)
        println("producing $i done")

从输出可以看出，虽然下游处理任务启动了，但是还没有完成的时候就被取消了（没有 consuming 2 done）：

producing 1
consuming 1
producing 1 done
producing 2
consuming 2
producing 2 done
producing 3
consuming 3
producing 3 done
consuming 3 done
all done
elapse: 509.858319ms

转换型

  • transform
  • map
  • filter
  • take
  • reduce

我们可以对流做一些转换操作，把发射出来的值转换成其他类型或值。比如最基本的 map()、filter() 操作符，这些操作符支持 suspend 方法，里面可以是耗时操作：

fun /*transformation_*/main() = runBlocking 
    (1..3).asFlow().map  request -> performRequest(request) 
        .collect  response -> println(response) 


suspend fun performRequest(request: Int): String 
    delay(1000)
    return "response $request"

输出为：

(delay 1 sec)
response 1
(delay 1 sec)
response 2
(delay 1 sec)
response 3

map()、filter()、transform() 都是 Flow 的扩展方法，map()和filter()本质上是调用 transform() 实现的，transform() 自己又构造了一个 flow：

public inline fun <T, R> Flow<T>.transform(
    @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = flow  // Note: safe flow is used here, because collector is exposed to transform on each operation
    collect  value ->
        // kludge, without it Unit will be returned and TCE won't kick in, KT-28938
        return@collect transform(value)

数量型

我们可以使用 take() 操作符只取前几个发射的值，后续的值将会被cancel掉，同时抛出异常：

fun /*sizeLimiting_*/main() = runBlocking 
    numbers()
        .take(2)
        .collect  value -> println(value) 


fun numbers() = flow 
    try 
        emit(1)
        emit(2)
        println("this line will not be executed")
        emit(3)
     catch (e: Throwable) 
        println(e.message)
     finally 
        println("finally in numbers")

输出：

1
2
Flow was aborted, no more elements needed
finally in numbers

启动型

collect() 也是一个操作符。除了 collect()，还有哪些操作符能启动一个异步流吗？有的，大概有如下几类：

  • 转成集合，比如 toList、toSet；
  • 获取第一个发射值以及保证只有一个发射值（无发射值或多于一个则会抛异常）的操作符，如 first()和single()；在这里插入代码片
  • 将所有发射值缩减至一个值的操作符，如 reduce()、fold()；
  • 新启协程运行异步流，比如 launchIn()；

这些操作符有如下特点：

  • 只能出现在异步流的最后，不能与 collect() 同时出现；
  • 都是 suspend 方法；
fun /*trigger_*/main() = runBlocking 
    val res = (1..5).asFlow()
        .map  it * it 
//        .single()
//        .first()
        .reduce  a, b -> a + b 
//        .fold(1)  acc: Int, value: Int -> acc + value 

    println(res)

输出：55

launchIn() 操作符比较特殊，使用它可以另外启动一个协程用于运行当前异步流。一般 collect() 方法启动的异步流会直接使用当前协程运行，但是就导致 collect() 方法后面的代码需要等待，等待异步流执行完成后才能开始执行。如果我们希望 collect() 方法立即执行，那么可以使用 launchIn()。比如使用 collect() 方法启动异步流时：

fun /*launch_*/main() = runBlocking 
    (1..3).asFlow().onEach  delay(100) .onEach  value -> println(value) /*.launchIn(this)*/.collect()
    println("DONE")

输出为：

1
2
3
DONE

显然异步流后面的代码是等待异步流执行完成之后才执行的。如果我们使用 launchIn()：

fun /*launch_*/main() = runBlocking 
    (1..3).asFlow().onEach  delay(100) .onEach  value -> println(value) .launchIn(this)/*.collect()*/
    println("DONE")

输出为：

DONE
1
2
3

launchIn() 会另外开启一个新的协程运行异步流的代码，与 printlin("DONE")是同时运行。其效果相当于：

fun /*launch_*/main() = runBlocking 
    launch  (1..3).asFlow().onEach  delay(100) .onEach  value -> println(value) .collect() 
    println("DONE")

组合型

  • zip
  • combine
  • flatten

zip() 可以把两个flow组合起来：

import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.zip
import kotlinx.coroutines.runBlocking

fun /*compose_*/main() = runBlocking 
    val keys = flowOf("one", "two", "three", "four").onEach  delay(300) 
    val values = (1..3).asFlow().onEach  delay(400) 

    keys.zip(values)  a, b -> "$a -> $b" .collect  println(it)

输出：

one -> 1
two -> 2
three -> 3

执行流程示意图：

combine()
combine 的作用是，每当其中一个流有最新值时，该最新值都会与其他流的最新值重新执行一遍处理流程。

fun /*compose_*/main() = runBlocking 
    val keys = flowOf("one", "two", "three", "four").onEach  delay(1000) 
    val values = (1..3).asFlow().onEach  delay(2000) 

    keys/*.zip*/.combine(values)  a, b -> "$a -> $b" .collect  value -> println("$value")

每隔2000ms输出一行：

one -> 1
two -> 1
three -> 1
three -> 2
four -> 2
four -> 3

执行流程为：

flatten
在实际开发中，有一种场景是一个flow发射的每一个值会使用map()操作符产生一个新的flow，于是就有了一个流组成的流（a flow of flows）：

fun /*flatten_*/main() = runBlocking 
    (1..3).asFlow()以上是关于对比Java学Kotlin协程-异步流的主要内容，如果未能解决你的问题，请参考以下文章
 

 对比Java学Kotlin协程-异步流


 对比Java学Kotlin协程-异步流


 对比Java学Kotlin协程-异步流


 对比Java学Kotlin协程-异步流


 对比Java学Kotlin协程简史


 对比Java学Kotlin协程简史