Kotlin（二十一）协程（异步流）

1.Flows（Flows）

挂起函数可以通过返回值返回单个计算值，但如何返回多个计算值呢？我们可以使用Flows。我们使用集合遍历，打印来返回多个计算值来举例子

package com.example.kotlin01
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
 fun main()  = runBlocking {
     launch {
        for (k in 1..3){
            println("我没有被阻塞，当前值$k")
            delay(100)
        }
     }
     foo().collect{
         value ->
         println(value)
     }
 }
fun foo(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}

我没有被阻塞，当前值1
1
我没有被阻塞，当前值2
2
我没有被阻塞，当前值3
3

从打印的结果来看，两个打印的地方都是异步的，并没有阻塞主线程。

Flow类型的构造器函数名为flow
flow中的代码块可以挂起
计算值通过emit函数发出
结果值通过collect函数从fllow取出
此场景有点类型RxJava

2.流是冷的（Flows are cold）

流是冷的意思是，flow build中的代码在开始收集之前不会运行，也就是在collect之前，不会运行flow里面的代码块

package com.example.kotlin01

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking


fun main() = runBlocking {
    var flow = foo()
    flow.collect { value ->
        println(value)
    }
    println("flow agin")
    flow.collect { value ->
        println(value)
    }
}


fun foo(): Flow<Int> = flow {
    println("flow start")
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}

flow start
1
2
3
flow agin
flow start
1
2
3

每次 collect的时候都会start

3.取消流（Flow cancellation）

流的收集操作可以在当流在一个可以取消的挂起函数中挂起的时候取消（比如delay ）,否则不能取消
下面通过withTimeoutOrNull 块演示流如何在超时的时候被取消并停止执行

package com.example.kotlin01

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull


fun main() = runBlocking {
    var flow = foo()
    withTimeoutOrNull(250) {
        flow.collect { value ->
            println("collect $value")
        }
    }
    println("done")


}


fun foo(): Flow<Int> = flow {

    for (i in 1..Int.MAX_VALUE) {
        delay(100)
        println("emit $i")
        emit(i)
    }
}

emit 1
collect 1
emit 2
done

以上的代码，foo函数中每隔100毫秒收集一次，main函数，收集flow的计算值。并且在250毫秒的时候超时。我们看打印知道，打印了1和2之后，就打印了 done,因为此时当前流在一个可以取消的挂起函数中delay函数挂起。因为超时之后被取消了。

4.流构建器（Flow builders）

前面说的flow{}是最基础的流构建器，还有其他构建器可以声明流

flowOf()定义了一个发射固定值集的流构建器
可以使用asFlow()扩展函数将各种集合和序列转换成流

fun main() = runBlocking {
    (1..3).asFlow().collect() { value -> println(value) }

}

1
2
3

5. 中间流运算符

可以使用运算符来转换流，中间运算符应用于上游流并返回下游流，比如我们熟悉的map，filter
使用map将请求值映射成结果值

import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.runBlocking


fun main() = runBlocking {
    (1..5).asFlow().map { value -> request(value) }
        .collect { response -> println(response) }


}
suspend fun request(parm: Int): Int {
    delay(100)
    return parm
}

5.1.转换操作符（Transform operator）

转换运算符Transform可以用来模拟简单的数据转换也可以用来更复杂的运算符转换
使用Transform模拟在执行一个耗时任务之前发出一个字符串并在字符串后跟出一个响应

package com.example.kotlin01

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.transform
import kotlinx.coroutines.runBlocking


fun main() = runBlocking<Unit> {
   val flow =  (1..5).asFlow().transform { param ->
        emit("这是第 $param 次请求")
        emit(request(param))
    }
    flow.collect{data-> println(data)}


}

suspend fun request(parm: Int): Int {
    delay(100)
    return parm
}

这是第 1 次请求
1
这是第 2 次请求
2
这是第 3 次请求
3
这是第 4 次请求
4
这是第 5 次请求
5

5.2.限长运算符（Size-limiting operators）

限长运算符在达到相应限制的时候，取消流的执行。协程中的取消总是通过抛出异常来实现的，这样所有的资源管理函数就可以在取消时正常执行
限制长度是2,我们可以看到打印的结果2之后就停止了

package com.example.kotlin01
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.runBlocking
fun main() = runBlocking<Unit> {
  try {
      val flow = (1..5).asFlow().take(2).map { value ->
          request(value)
      }.collect { response -> println(response) }
  } finally {
      println("Finally in numbers")

  }
}
suspend fun request(parm: Int): Int {
  delay(100)
  return parm
}

1
2
Finally in numbers

6. 流运算符（Terminal flow operators）

终端流运算符是用于启动流的挂起函数，collect是最基本的终端流运算符，但还要其他的终端流运算符，可以使得操作更加简单

转换各种集合 toList，toSet
first 用于获取第一个值，single，用于确保流发出单个值
使用reduce和fold 将流还原为某个值

package com.example.kotlin01

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking


fun main() = runBlocking<Unit> {
    try {
        val list = (1..5).asFlow().take(2).map { value ->
            request(value)
        }.toList()
        println(list)
    } finally {
        println("Finally in numbers")

    }


}

suspend fun request(parm: Int): Int {
    delay(100)
    return parm
}

[1, 2]
Finally in numbers

将流转换成集合

7. 流是连续的（Flows are sequential）

每个流的单独集合都是按照顺序执行的。

package com.example.kotlin01

import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.runBlocking


fun main() = runBlocking<Unit> {
    (1..5).asFlow().filter {
        println("当前传入的值$it")
        it % 2 == 0//过滤下 模 2 = 0的
    }
        .map {
            println("当前转换的值$it")
            "$it"
        }
        .collect {
            println("当前收集的值 $it ")
        }

}

当前传入的值1
当前传入的值2
当前转换的值2
当前收集的值 2 
当前传入的值3
当前传入的值4
当前转换的值4
当前收集的值 4 
当前传入的值5

只有2和4是满足的，2和4会执行到收集完毕，再继续发射下一个值。

8. 流上下文

流的收集总是在调用协程的上下文中执行

package com.example.kotlin01

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking


fun main() = runBlocking {
    foo().collect { value -> println("当前线程${Thread.currentThread().name}-当前值$value") }

}

fun foo(): Flow<Int> = flow {
    for (i in 1..4) {
        emit(i)
    }

}

当前线程main-当前值1
当前线程main-当前值2
当前线程main-当前值3
当前线程main-当前值4

调用协程的地方是在主线程中执行的，所有流的收集也是在主线程中执行。因为其上下文在主线程。

8.1.错误地使用 withContext（Wrong emission withContext）

通常withContext用于使用协程的时候更改代码中的上下文，但是flow中的代码块必须要保留上下文的属性，不能修改上下文。

package com.example.kotlin01

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext


fun main() = runBlocking {
    foo().collect { value -> println("当前线程${Thread.currentThread().name}-当前值$value") }

}

fun foo(): Flow<Int> = flow {
    withContext(Dispatchers.Default){
        for (i in 1..4) {
            emit(i)
        }
    }


}

Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated:
        Flow was collected in [BlockingCoroutine{Active}@614b37e6, BlockingEventLoop@185a864f],
        but emission happened in [DispatchedCoroutine{Active}@64e54ce0, Dispatchers.Default].
        Please refer to 'flow' documentation or use 'flowOn' instead
    at kotlinx.coroutines.flow.internal.SafeCollector_commonKt.checkContext(SafeCollector.common.kt:84)
    at kotlinx.coroutines.flow.internal.SafeCollector.checkContext(SafeCollector.kt:84)
    at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:70)
    at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:55)
    at com.example.kotlin01.CoroutineKt$foo$1$1.invokeSuspend(coroutine.kt:19)

8.2.flowOn 运算符（flowOn operator）

有个例外就是flowOn运算符可以修改发射时的上下文


import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.runBlocking


fun main() = runBlocking {
    foo().collect { value -> println("当前线程${Thread.currentThread().name}-当前值$value") }

}

fun foo(): Flow<Int> = flow {
        for (i in 1..4) {
            Thread.sleep(100)
            emit(i)
            println("发射值的当前线程${Thread.currentThread().name}")
        }
}.flowOn(Dispatchers.Default)

发射值的当前线程DefaultDispatcher-worker-1
当前线程main-当前值1
发射值的当前线程DefaultDispatcher-worker-1
当前线程main-当前值2
发射值的当前线程DefaultDispatcher-worker-1
当前线程main-当前值3
发射值的当前线程DefaultDispatcher-worker-1
当前线程main-当前值4

结果是发射时的线程是在DefaultDispatcher，而收集值的线程是在main

9.缓冲（Buffering）

package com.example.kotlin01

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlin.system.measureTimeMillis


fun main() = runBlocking {
    val time = measureTimeMillis {
        foo().collect {
                value ->
            delay(300)
        }
    }
    println("当前流执行需要花费的时间$time")

}

fun foo(): Flow<Int> = flow {
    for (i in 1..10) {
        delay(100)
        emit(i)
    }
}

当前流执行需要花费的时间4081

以上的代码我们是按照执行，先执行完收集代码，在执行fool.时间花费是4081

package com.example.kotlin01

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlin.system.measureTimeMillis


fun main() = runBlocking {
    val time = measureTimeMillis {
        foo().buffer().collect {
                value ->
            delay(300)
        }
    }
    println("当前流执行需要花费的时间$time")

}

fun foo(): Flow<Int> = flow {
    for (i in 1..10) {
        delay(100)
        emit(i)
    }
}

当前流执行需要花费的时间3214

使用buffer运算符在运行取值代码的同时同时运行 fool(),而不是按顺序运行他门.时间花费了3214

9.1合并（Conflation）

有时候我们的需求是不关心每一个值的结果,值关心最近的值,比如最新的状态等.当时可能前面的值还在处理中,导致没法生成后面的值,这样我们可以使用Conflation跳过前面的值

package com.example.kotlin01

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlin.system.measureTimeMillis


fun main() = runBlocking {
    val time = measureTimeMillis {
        foo().conflate().collect {
                value ->
            delay(300)
            println(value)
        }
    }
    println("当前流执行需要花费的时间$time")

}

fun foo(): Flow<Int> = flow {
    for (i in 1..10) {
        delay(100)
        emit(i)
    }
}

我们看到前面没有完全打印,比如4,是因为1这个数字还在处理中,而234已经生成,所以就可以合并到4中.所以打印了4

9.2.处理最新值（Processing the latest value）

在发射端和收集端处理很慢的时候,合并加快处理速度的一种方法,如9.1,她通过丢弃发射值来实现.另外一种方式是取消慢速的收集器,每次发送出新值的时候,重新启动她.比如我们可以把Conflation改成collectLatest.字面意思上就是收集最近的发射值

package com.example.kotlin01

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlin.system.measureTimeMillis


fun main() = runBlocking {
    val time = measureTimeMillis {
        foo().collectLatest { value ->
            delay(300)
            println(value)
        }
    }
    println("当前流执行需要花费的时间$time")

}

fun foo(): Flow<Int> = flow {
    for (i in 1..10) {
        delay(100)
        emit(i)
    }
}

10
当前流执行需要花费的时间1467

10.组合多个流（Composing multiple flows）

10.1.zip

和kotlin标准库中的 Sequence.zip 扩展函数一样,流有一个zip合并运算符,组合两个流的相应的值

package com.example.kotlin01

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 main() = runBlocking {
    val numbers = (1..3).asFlow()
    val strs = flowOf("a", "b", "c", "d")
    numbers.zip(strs) { a, b ->"$a and $b" }.collect { value ->
        println(value)
    }


}

1 and a
2 and b
3 and c

image.png

从上面我们知道两个流是对应合并
10.2.Combine
合并最近的

package com.example.kotlin01

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking


fun main() = runBlocking {
    val numbers = (1..3).asFlow()
    val strs = flowOf("a", "b", "c")
    numbers.combine(strs) { a, b ->"$a and $b" }.collect { value ->
        println("$value")

    }

}

1 and a
2 and a
2 and b
3 and b
3 and c

image.png

(1) 1和a - >1a
(2) a和2 ->2a
(3) 2和b ->2b
(4) b和3 ->3b
(5) 3和c ->3c

11.展平流（Flattening flows）

package com.example.kotlin01

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking


fun main() = runBlocking<Unit> {
    //map运算符将请求值映射成结果值
    //传入的是 1 2 3，得到的结果是一个flow  <flow<String>>
    (1..3).asFlow().map { request(it) }

}

fun request(request: Int): Flow<String> = flow {
    emit("这是第一个 $request")
    delay(500)
    emit("这是第二个 $request")
}

我们看以上的代码,可以看到,main函数中的代码块得到的是一个flow <flow<String>>流.也就是流中包含了流.所以我们需要将其展平为单独的一个流进行处理.

11.1.flatMapConcat

使用flatMapConcat操作符.,等待内部完成时,然后收集下一个流

package com.example.kotlin01

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking


fun main() = runBlocking<Unit> {
    //map运算符将请求值映射成结果值
    //传入的是 1 2 3，得到的结果是一个flow  <flow<String>>
    (1..3).asFlow().flatMapConcat { request(it) }.collect{ println(it)}
}

fun request(request: Int): Flow<String> = flow {
    emit("这是第一个 $request")
//    delay(500)
//    emit("这是第二个 $request")
}

这是第一个 1
这是第一个 2
这是第一个 3

11.2.flatMapMerge

flatMapMerge操作符模式是同时收集传入的流,然后合并到单个流中.

package com.example.kotlin01

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking


fun main() = runBlocking<Unit> {
    //map运算符将请求值映射成结果值
    //传入的是 1 2 3，得到的结果是一个flow  <flow<String>>
    //concurrency参数指的是限制同时收集的并发数
    //(1..3).asFlow().flatMapMerge(100, { request(it) }).collect { println(it) }
    (1..3).asFlow().flatMapMerge { request(it)  }.collect{ println(it)}
}

fun request(request: Int): Flow<String> = flow {
    emit("这是第一个 $request")
//    delay(500)
//    emit("这是第二个 $request")
}

这是第一个 1
这是第一个 2
这是第一个 3

11.3.flatMapLatest

和之前章节的collectLatest 操作符类似,一旦有新流发出,将取消之前已经发出的流.

package com.example.kotlin01

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking


fun main() = runBlocking<Unit> {
    //map运算符将请求值映射成结果值
    //传入的是 1 2 3，得到的结果是一个flow  <flow<String>>
    (1..3).asFlow().flatMapLatest { request(it)  }.collect{ println(it)}
}

fun request(request: Int): Flow<String> = flow {
    emit("这是第一个 $request")
//    delay(500)
//    emit("这是第二个 $request")
}

这是第一个 1
这是第一个 2
这是第一个 3

以上我们通过三个操作符都可以将流展平.

12.流异常（Flow exceptions）

当发射器或运算符内部的代码引发异常的时候,流收集器可以结束运行.但会出现异常.可以通过 try catch

fun main() = runBlocking<Unit> {
    //map运算符将请求值映射成结果值
    //传入的是 1 2 3，得到的结果是一个flow  <flow<String>>
    try {
        (1..3).asFlow().collect {
            println(it)
            it / 0
        }
    }
    catch (e:Exception){
        println(e.message)
    }

}

13.异常透明性

使用catch运算符捕捉异常,catch只能收集下游的异常,如果上游的代码出现异常就不会收集.

package com.example.kotlin01

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import java.lang.Exception


fun main() = runBlocking<Unit> {
    request(2).catch { println("这是异常...$it") }
        .collect {
            println(it)
        }
}

fun request(request: Int): Flow<String> = flow {
    emit("这是第一个 $request")
    request/0

}

这是第一个 2
这是异常...java.lang.ArithmeticException: / by zero

14.流完成（Flow completion）

流程在完成的之后(正常完成或者异常完成)可以做一些操作,可以通过两种方式完成,命令式和声明式

14.1.命令式

使用try/finally

package com.example.kotlin01

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import java.lang.Exception


fun main() = runBlocking<Unit> {
    try {
        request(2).catch { println("这是异常...$it") }
            .collect {
                println(it)
            }
    }
    finally {
        println("done")
    }

}

fun request(request: Int): Flow<String> = flow {
    emit("这是第一个 $request")
    request/0

}

这是第一个 2
这是异常...java.lang.ArithmeticException: / by zero
done

14.2.声明式

使用onCompletion 操作符

package com.example.kotlin01

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import java.lang.Exception


fun main() = runBlocking<Unit> {

    request(2).onCompletion { println("done") }.catch { println("这是异常...$it") }
        .collect {
            println(it)
        }
}

fun request(request: Int): Flow<String> = flow {
    emit("这是第一个 $request")
}

这是第一个 2
done

十五、命令式还是声明式

我们知道如何收集流,并以命令和声明式的处理完成和异常.那到底选择哪一种呢,根据编程爱好来,我觉得使用声明式比较舒服

十六、启动流（Launching flow)

我们可以使用launchIn代替collect收集流,launchIn的作用是在单独协程中启动收集流.而collect不是.

package com.example.kotlin01

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking


fun main() = runBlocking<Unit> {

    request(2)
        //onEach 是一个中间运算符
        .onEach { println(it) }
        //单独启动收集流的协程进行对流的收集
        //.collect()
        .launchIn(this)
}

fun request(request: Int): Flow<String> = flow {
    emit("这是第一个 $request")

}


这是第一个 2