Kotlin Flow使用

2022-12-16 本文已影响0人因为我的心

一、前言：

Flow是kotlin提供的一个工具，使用协程封装成生产者-消费者模式，上流来负责生产，下流来接收消耗。

A cold asynchronous data stream that sequentially emits values
and completes normally or with an exception。
翻译下就是：按顺序发出值并正常完成或异常完成的Cold异步数据流。

热流: 无论有没有 Subscriber 订阅，事件始终都会发生。当 Hot Observable 有多个订阅者时，Hot Observable 与订阅者们的关系是一对多的关系，可以与多个订阅者共享信息。
冷流: 只有 Subscriber 订阅时，才开始执行发射数据流的代码。并且 Cold Observable 和 Subscriber 只能是一对一的关系，当有多个不同的订阅者时，消息是重新完整发送的。也就是说对 Cold Observable 而言，有多个Subscriber的时候，他们各自的事件是独立的。

image.png

二、Flow使用：

1、Flow 的简单使用

（1）flow{ ... }内部可以调用suspend 函数；
（2）使用 emit() 方法来发射数据；
（3）使用 collect() 方法来收集结果。

1、Flow 的简单使用

//上流函数
fun simpleFlow() = flow {
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}
 
fun main() {
    runBlocking {
        //下流接收数据
        simpleFlow().collect { value ->
            println(value)
        }
 
        println("finished")
    }
}

结果：
1
2
3
finished

2、Flow是冷流，所以collect是挂起函数，不是子协程，并且只有执行collect函数时，上流的代码才会被执行，所以在一个协程中多次调用collect，它们会按顺序执行。

fun simpleFlow() = flow {
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}
 
fun main() {
    runBlocking {
        simpleFlow().collect { value ->
            println(value)
        }
 
        println("collect1 finished")
 
        simpleFlow().collect { value ->
            println(value)
        }
 
        println("collect2 finished")
    }
}

结果：
1
2
3
collect1 finished
1
2
3
collect2 finished

3、Flow的连续性

Flow也支持函数式编程，并且从上流到下流的每个过渡操作符都会处理发射值，最终流入下流

fun main() {
    runBlocking {
        flow {
            for (i in 1..5) {
                delay(100)
                emit(i)
            }
        }.filter {
            it % 2 == 0 //只取偶数
        }.map {
            "String $it"
        }.collect {
            println(it)
        }
    }
}
结果：
String 2
String 4

2、创建 Flow 的常用方式

（1）flow{ ... }需要显示调用emit() 发射数据；
（2）flowOf()一个发射固定流，不需要显示调用emit() 发射数据；
（3）asFlow()扩展函数，可以将各种集合与序列转换为流，也不需要显示调用emit() 发射数据；

1、flow{}

flow {
    (5 .. 10).forEach {
              emit(it)
         }
}.collect{
   println(it)
}

2、flowOf() 帮助可变数组生成 Flow 实例

flowOf(1,2,3,4,5).collect { println(it) }

其实flowOf调用的就是第一种flow{}，分别emit发送值，源码如下:

public fun <T> flowOf(vararg elements: T): Flow<T> = flow {
    for (element in elements) {
        emit(element)
    }
}

3、asFlow() 面向数组、列表等集合

 (1 ..5).asFlow().onEach {
        delay(1000)
  }.collect {
         println(it)
  }

4、注意：消费数据
collect 方法和 RxJava 中的 subscribe 方法一样，都是用来消费数据的。
除了简单的用法外，这里有两个问题得注意一下：

collect 函数是一个 suspend 方法，所以它必须发生在协程或者带有 suspend 的方法里面，这也是我为什么在一开始的时候启动了

lifecycleScope.launch。lifecycleScope 是我使用的 Lifecycle 的协程扩展库当中的，你可以替换成自定义的协程作用域

三、Flow切换线程：

1、切换线程使用的是flowOn操作符。

    flow {
        for (i in 1..5) {
            delay(100)
            emit(i)
        }
    }.map {
        it * it
    }.flowOn(Dispatchers.IO)
        .collect {
            println(it)
        }

简单点理解:就是flowOn之前的操作符运行在flowOn指定的线程之内，flowOn之后的操作符运行在整个flow运行的CoroutineContext内。

例如，下面的代码collect则是在main线程:

fun main() = runBlocking {
  flowOf(1,2,3,4,5)
                    .flowOn(Dispatchers.Default)
                    .collect { 
                    println(Thread.currentThread().name+" "+it) 
                }
}

打印如下:

main 1
main 2
main 3
main 4
main 5

2、除了使用子协程执行上流外，我们还可以使用launchIn函数来让Flow使用全新的协程上下文。

public fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job = scope.launch {
    collect() // tail-call
}

fun main() {
    runBlocking {
        flow {
            println("flow ：${Thread.currentThread().name}")
            for (i in 1..5) {
                delay(100)
                emit(i)
            }
        }.flowOn(Dispatchers.Default)
            .onEach { println("collect：${Thread.currentThread().name} $it") }
            .launchIn(CoroutineScope(Dispatchers.IO))
            .join()//主线程等待这个协程执行结束
    }
}
结果：
flow ：DefaultDispatcher-worker-1
collect：DefaultDispatcher-worker-1 1
collect：DefaultDispatcher-worker-1 2
collect：DefaultDispatcher-worker-1 3
collect：DefaultDispatcher-worker-1 4
collect：DefaultDispatcher-worker-1 5

四、Flow操作符

1、transform

在使用transform操作符时，可以任意多次调用emit。

runBlocking {

                (1..5).asFlow()
                    .transform {
                        emit(it * 2)
                        delay(100)
                        emit(it * 4)
                    }
                    .collect { println("transform:$it") }
            }

打印如下:

transform:2
transform:4
transform:4
transform:8
transform:6
transform:12
transform:8
transform:16
transform:10
transform:20

transform、transformLatest、transformWhile ，transform直接进行转换，和map不同的是transform可以控制流速，transformLatest则进行最新值的转换，类似于mapLatest ，transformWhile则要求闭包返回一个boolean值，为true则继续返回，为false则后续的值全部取消。

日志如下：

2022-07-29 15:37:03.243 10589-10615/edu.test.demo D/Test-TAG: transform is 0
2022-07-29 15:37:04.255 10589-10615/edu.test.demo D/Test-TAG: transform is 10
2022-07-29 15:37:05.269 10589-10615/edu.test.demo D/Test-TAG: transform is 20
2022-07-29 15:37:06.281 10589-10615/edu.test.demo D/Test-TAG: transform is 30
2022-07-29 15:37:07.294 10589-10615/edu.test.demo D/Test-TAG: transform is 40
2022-07-29 15:37:08.306 10589-10615/edu.test.demo D/Test-TAG: transform is 50
2022-07-29 15:37:09.318 10589-10615/edu.test.demo D/Test-TAG: transform is 60
2022-07-29 15:37:10.330 10589-10615/edu.test.demo D/Test-TAG: transform is 70
2022-07-29 15:37:11.341 10589-10615/edu.test.demo D/Test-TAG: transform is 80
2022-07-29 15:37:12.353 10589-10615/edu.test.demo D/Test-TAG: transform is 90
2022-07-29 15:37:13.470 10589-10617/edu.test.demo D/Test-TAG: transformLatest 9
2022-07-29 15:37:13.483 10589-10617/edu.test.demo D/Test-TAG: transformWhile 0
2022-07-29 15:37:13.495 10589-10617/edu.test.demo D/Test-TAG: transformWhile 1
2022-07-29 15:37:13.509 10589-10617/edu.test.demo D/Test-TAG: transformWhile 2
2022-07-29 15:37:13.521 10589-10617/edu.test.demo D/Test-TAG: transformWhile 3
2022-07-29 15:37:13.532 10589-10617/edu.test.demo D/Test-TAG: transformWhile 4
2022-07-29 15:37:13.544 10589-10617/edu.test.demo D/Test-TAG: transformWhile 5

2、take

take操作符只取前几个emit发射。

  (1 .. 5).asFlow().take(2).collect {
                    println("take:$it")
                }

打印结果:

take:1
take:2

take、takeWhile、drop、dropWhile，take则是取几个值返回，takeWhile按条件取值，如果满足条件就返回，不满足则后面全部取消。drop和take相反，dropWhile和takeWhile相反。

            val flow = flow {
                repeat(10){
                    delay(10)
                    emit(it)
                }
            }
              flow.take(5).collect {
                Log.d(TAG.TAG,"take $it")
            }

            flow.takeWhile {
                it < 5
            }.collect {
                Log.d(TAG.TAG,"takeWhile $it")
            }

            flow.drop(5).collect {
                Log.d(TAG.TAG,"drop $it")
            }
            flow.dropWhile {
                it < 5
            }.collect {
                Log.d(TAG.TAG,"dropWhile $it")
            }

打印如下：

D/Test-TAG: take 0
D/Test-TAG: take 1
D/Test-TAG: take 2
D/Test-TAG: take 3
D/Test-TAG: take 4
D/Test-TAG: takeWhile 0
D/Test-TAG: takeWhile 1
D/Test-TAG: takeWhile 2
D/Test-TAG: takeWhile 3
D/Test-TAG: takeWhile 4
D/Test-TAG: drop 5
D/Test-TAG: drop 6
D/Test-TAG: drop 7
D/Test-TAG: drop 8
D/Test-TAG: drop 9
D/Test-TAG: dropWhile 5
D/Test-TAG: dropWhile 6
D/Test-TAG: dropWhile 7
D/Test-TAG: dropWhile 8
D/Test-TAG: dropWhile 9

3、reduce

runBlocking {
                val sum=( 1 ..5).asFlow()
//                    .map {
//                    //println("map:${it}")
//                    it*it  }   //1,4,9,16,25

                    .reduce { a, b ->
                        println("reduce:${a},${b}")
                        a*b
                    }

                 println(sum)

            }

打印如下:

reduce:1,2
reduce:2,3
reduce:6,4
reduce:24,5
120

简单点理解就是两个元素操作之后拿到的值跟后面的元素进行操作,用于把flow 简化合并为一个值。

4、zip

fun main() = runBlocking {

    val flowA = (1..5).asFlow()
    val flowB = flowOf("one", "two", "three", "four", "five").onEach { delay(100) }

    val time = measureTimeMillis {
        flowA.zip(flowB) { a, b -> "$a and $b" }
            .collect { println(it) }
    }

    println("Cost $time ms")
}

打印如下:

1 and one
2 and two
3 and three
4 and four
5 and five
Cost 540 ms

如果flowA中的item个数大于flowB中的item个数，执行合并后新flow的item个数=较小的flow的item个数。

5、filter

filter、filterNot、filterIsInstance、filterNotNull、fliter闭包返回一个Boolean值，为true则返回，false则不返回，filterNot刚好相反；filterIsInstance则进行类型过滤，如过滤出String或者Int等，filterNotNull则过滤null值，返回非空值。

            val flow = flow {
                repeat(10){
                    delay(10)
                    emit(it)
                }
            }
            flow.filter {
                it % 2 == 0
            }.collect {
                Log.d(TAG.TAG,"filter $it")
            }

            flow.filterNot {
                it % 2 == 0
            }.collect {
                Log.d(TAG.TAG,"filterNot $it")
            }

            flow {
                emit(1)
                emit("123")
            }.filterIsInstance<String>().collect {
                Log.d(TAG.TAG,"filterIsInstance $it")
            }

            flow {
                emit(1)
                emit(null)
                emit(2)
            }.filterNotNull().collect {
                Log.d(TAG.TAG,"filterNotNull $it")
            }

打印如下：

2022-07-29 15:50:45.376 10675-10703/edu.test.demo D/Test-TAG: filter 0
2022-07-29 15:50:45.400 10675-10703/edu.test.demo D/Test-TAG: filter 2
2022-07-29 15:50:45.422 10675-10703/edu.test.demo D/Test-TAG: filter 4
2022-07-29 15:50:45.444 10675-10703/edu.test.demo D/Test-TAG: filter 6
2022-07-29 15:50:45.466 10675-10703/edu.test.demo D/Test-TAG: filter 8
2022-07-29 15:50:45.505 10675-10703/edu.test.demo D/Test-TAG: filterNot 1
2022-07-29 15:50:45.528 10675-10703/edu.test.demo D/Test-TAG: filterNot 3
2022-07-29 15:50:45.550 10675-10703/edu.test.demo D/Test-TAG: filterNot 5
2022-07-29 15:50:45.574 10675-10703/edu.test.demo D/Test-TAG: filterNot 7
2022-07-29 15:50:45.597 10675-10703/edu.test.demo D/Test-TAG: filterNot 9
2022-07-29 15:50:45.598 10675-10703/edu.test.demo D/Test-TAG: filterIsInstance 123
2022-07-29 15:50:45.600 10675-10703/edu.test.demo D/Test-TAG: filterNotNull 1
2022-07-29 15:50:45.600 10675-10703/edu.test.demo D/Test-TAG: filterNotNull 2

6、merge

是将两个flow合并起来，将每个值依次发出来

            val flow1  = listOf(1,2).asFlow()
            val flow2 = listOf("one","two","three").asFlow()
            merge(flow1,flow2).collect {value->
                Log.d(TAG.TAG,value.toString())
            }

可以看出merge在将flow1和flow2合并之后将五个值依次发送出来。

五、Flow的异常处理

当运算符中的发射器或代码抛出异常，可以有两种方式处理
1.try catch
2.catch函数

1、try catch适用于收集时发生的异常

fun main() {
    runBlocking {
        val flow = flow {
            for (i in 1..3) {
                emit(i)
            }
        }
 
        try {
            flow.collect {
                println(it)
                throw RuntimeException()
            }
        } catch (e: Exception) {
            print("caught: $e")
        }
    }
}

2、虽然上流也可以使用try catch，但是更推荐catch函数

fun main() {
    runBlocking {
        val flow = flow {
            for (i in 1..3) {
                emit(i)
                throw RuntimeException()
            }
        }.catch { e ->
            print("caught1: $e")
        }.collect {
            println(it)
        }
    }
}

六、Flow的完成

  runBlocking {
            val flow = flow {
                for (i in 1..3) {
                    emit(i)
                }
            }.onStart {
                //1、开始操作符
                println("start")
            }.onCompletion {
                //1、完成操作符
                println("done")
            }.collect {
                println(it)
            }
        }

七、取消Flow

Flow也是可以被取消的，最常用的方式就是通过withTimeoutOrNull来取消，代码如下所示。

MainScope().launch {
    withTimeoutOrNull(2500) {
        flow {
            for (i in 1..5) {
                delay(1000)
                emit(i)
            }
        }.collect {
            Log.d("xys", "Flow: $it")
        }
    }
}

这样当输出1、2之后，Flow就被取消了。

Flow的取消，实际上就是依赖于协程的取消。

八、Flow的同步非阻塞模型

首先，我们要理解下，什么叫同步非阻塞，默认场景下，Flow在没有切换线程的时候，运行在协程作用域指定的线程，这就是同步，那么非阻塞又是什么呢？我们知道emit和collect都是suspend函数，所谓suspend函数，就是会挂起，将CPU资源让出去，这就是非阻塞，因为suspend了就可以让一让，让给谁呢？让给其它需要执行的函数，执行完毕后，再把资源还给我。

flow {
    for (i in 0..3) {
        emit(i)
    }
}.onStart {
    Log.d("xys", "Start Flow in ${Thread.currentThread().name}")
}.onEach {
    Log.d("xys", "emit value---$it")
}.collect {
    Log.d("xys", "Result---$it")
}

输出为：

D/xys: Start Flow in main
D/xys: emit value---0
D/xys: Result---0
D/xys: emit value---1
D/xys: Result---1
D/xys: emit value---2
D/xys: Result---2
D/xys: emit value---3
D/xys: Result---3

可以发现，emit一个，collect拿一个，这就是同步非阻塞，互相谦让，这样谁都可以执行，看上去flow中的代码和collect中的代码，就是同步执行的。

九、异步非阻塞模型

假如我们给Flow增加一个线程切换，让Flow执行在子线程，同样是上面的代码，我们再来看下执行情况

flow {
    for (i in 0..3) {
        emit(i)
    }
}.onStart {
    Log.d("xys", "Start Flow in ${Thread.currentThread().name}")
}.onEach {
    Log.d("xys", "emit value---$it")
}.flowOn(Dispatchers.IO).collect {
    Log.d("xys", "Collect Flow in ${Thread.currentThread().name}")
    Log.d("xys", "Result---$it")
}

输出为：

D/xys: Start Flow in DefaultDispatcher-worker-1
D/xys: emit value---0
D/xys: emit value---1
D/xys: emit value---2
D/xys: emit value---3
D/xys: Collect Flow in main
D/xys: Result---0
D/xys: Collect Flow in main
D/xys: Result---1
D/xys: Collect Flow in main
D/xys: Result---2
D/xys: Collect Flow in main
D/xys: Result---3

D/xys: Result---3
这个时候，Flow就变成了异步非阻塞模型，异步呢，就更好理解了，因为在不同线程，而此时的非阻塞，就没什么意义了，由于flow代码先执行，而这里的代码由于没有delay，所以是同步执行的，执行的同时，collect在主线程进行监听。

除了使用flowOn来切换线程，使用channelFlow也可以实现异步非阻塞模型。

参考：Kotlin Flow 介绍 - 简书 (jianshu.com)

Kotlin语言（十三）：Flow - 简书 (jianshu.com)

Kotlin Flow使用

一、前言：

二、Flow使用：

1、Flow 的简单使用

2、创建 Flow 的常用方式

三、Flow切换线程：

1、切换线程使用的是flowOn操作符。

2、除了使用子协程执行上流外，我们还可以使用launchIn函数来让Flow使用全新的协程上下文。

四、Flow操作符

1、transform

2、take

3、reduce

4、zip

5、filter

6、merge

五、Flow的异常处理

1、try catch适用于收集时发生的异常

2、虽然上流也可以使用try catch，但是更推荐catch函数

六、Flow的完成

七、取消Flow

八、Flow的同步非阻塞模型

九、异步非阻塞模型

猜你喜欢

热点阅读