devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Kotlin 流flow、ShareFlow、StateFlow、Channel的解释与使用

作者：我家自动化 | 2024-08-19 00:39:06

赞

踩

Kotlin 流flow、ShareFlow、StateFlow、Channel的解释与使用

一、介绍

        随着Android接入kotlin开发，Android之前好多模式也渐渐被kotlin替代。开发模式也在做渐进的转型，从MVC到MVP在到MVVP以及现在的MVI等。

        流IO在java中和kotlin中使用率都是比较高的，场景很多。如Java的IO和NIO，再到我们现在使用的RxJava或者升级版的JxJava2。针对JxJava的流处理与转发机制。kotlin语言也自带了flow。

        早MVVM流行的时候，LiveData充当了数据的订阅与流转，但是我们知道，LiveData有一个问题，当Dialog或者当前设置发生旋转，会出现倒灌，这种行为是因为订阅生命发生变化。

针对这些问题，kotlin语言自带了flow流。

二、kotlin flow的介绍

        flow其实与RxJava比较类似，都是基于流在处理订阅的分发。这个flow也被最新的MVI模式所接受，替换了MVVM中的LiveData。

flow又区分冷流和热流。

冷流（Cold Flow）

  •  

    定义与工作原理：冷流是一种异步数据流，它按顺序发出值并正常或异常完成。冷流的特点是，只有当数据被消费者订阅（即调用collect方法后），生产者才开始执行发送数据流的代码。这意味着，如果没有消费者订阅，生产者不会发送任何数据。冷流通常是一对一的关系，即一个生产者对应一个消费者。

  •  

    创建方式：冷流可以通过多种方式创建，包括使用flow{}构建器、flowOf、asFlow等方法。这些方法允许开发者根据需要创建冷流，以满足特定的数据生产需求。

  •  

    适用场景：冷流适用于那些不需要持续生产数据，而是在数据被消费时才进行生产的场景。例如，从数据库查询数据、计算结果等，这些操作只有在需要时才会执行。

热流（Hot Flow）

  •  

    定义与工作原理：热流与冷流的主要区别在于，无论是否有消费者订阅，生产者都会持续发送数据。热流支持多个消费者同时订阅，并且生产的数据可以被多个消费者共享。热流通过stateIn和shareIn操作符实现，可以将任何流转化为热流。

  •  

    适用场景：热流适用于需要持续提供数据给多个消费者的场景，如实时数据更新、传感器数据读取等。由于热流能够同时支持多个消费者，因此在需要共享数据给多个组件或界面时非常有用。

通过冷热流区分我们可以知道冷流和热流可以相互转换。冷流是一对一服务，热流是一对多服务。

这个也就导致，在后期的开发过程中，订阅和消费，流的互相转换。

1对多

1对1

 三、流的使用

        接下来主要介绍Flow、SateFlow、ShareFlow以及Channel的用法，冷流和热流互换。这些流基本都是用到了协程，所以不能直接使用。

3.1.Flow

flow：这个老六就是热流，只要你订阅就一直发，不停的发，只要emit中的都有记录，而且内部还维持了一个index，可以知道目前下发了多少次。热流不管是什么视乎订阅，都会有。

场景：热流是无线转发，一般使用在状态订阅转发，比如平台关闭了id，可以直接放进去，后面其他模块如果可以订阅，查找当前状态。包括配合webscoket可以做很多事情。

  1.    var count=0
  2.             val myflow= flow<String> {
  3.                 repeat(3,{
  4.                     count++
  5.                     emit("当前是=${count}")
  6.                 })
  7.             }
  8.  
  9.  
  10.  
  11.             /**
  12.              * flow是一直保持的，只要订阅一次，就会发送一次，而且上次的记录一直保持。collect需要通过协程来处理
  13.              * */
  14.  
  15.             lifecycleScope.launch {
  16.                 myflow.flowOn(Dispatchers.Main).collect({
  17.                     textview.append("collect="+it+"\n")
  18.                 })
  19.                 myflow.collectIndexed { index, value ->
  20.                     textview.append("collectIndexed="+value+"，${index}\n")
  21.                 }
  22.                 myflow.flatMapConcat { it->
  23.                     flow {
  24.                         emit("map=${it}")
  25.                     }
  26.                 }.collect({
  27.                     textview.append("${it}\n")
  28.                 })
  29.  
  30.                 myflow.flatMapMerge {
  31.                     flow { emit("flatMapMerge=${it}")
  32.                         emit("MapMerge=最近")
  33.                     }
  34.                 }.collect({
  35.                     textview.append("${it}\n")
  36.                 })
  37.  
  38.                 myflow.flatMapLatest {
  39.                     flow { emit("flatMapLatest=${it}")
  40.                         emit("flatMapLatest=最近")
  41.                     }
  42.                 }.collect({
  43.                     textview.append("${it}\n")
  44.                 })
  45.                    //热流转冷
  46.  
  47.                 val reciver=myflow.produceIn(this).receiveCatching()
  48.                 textview.append("produceIn=${reciver.getOrNull()}\n")
  49.  
  50.             }

3.2 SateFlow

        状态流属于冷流，订阅会收到最近一次，如果订阅没有设置value，接受的是默认值，如果当前值没有变化，也不会发送，也就是当前值和上次会进行比较，如果一样不会继续订阅。是一对一，多次订阅，第一个collect完，后面将不会再收到订阅事件。

  1.         val sateFlow=MutableStateFlow("")
  2.  
  3.         var num=0
  4.         findViewById<View>(R.id.btn_click_state)?.setOnClickListener {
  5.             lifecycleScope.launch {
  6.  
  7.             //默认连接的时候会收到最近一次的，如果最近都没有设置value或者emit，那么收到的是默认值
  8.  
  9.                 sateFlow.value="ssssss"
  10.  
  11.                 sateFlow.value="ssssss111111111111=${num++}"
  12.  
  13.             }
  14.         }
  15.  
  16.  
  17.         lifecycleScope.launch {
  18.  
  19.  
  20.             sateFlow.collectIndexed { index, value ->
  21.                 textview.append("index=${index},value=${value}\n")
  22.             }
  23.             sateFlow.collect({
  24.                 textview.append("${it}\n")
  25.             })
  26.             sateFlow.emit("aaaa")
  27.         }

3.3 ShareFlow

        shareflow和其他的flow不一样，在构造的时候有三个参数，第一个是replay重复多少次，和缓存次数，后面是缓存流类型，内部也是可以多次订阅，但是collectIndexed和collect类型只能出现一个。但是一个类型可以多次订阅

  1. public fun <T> MutableSharedFlow(
  2.     replay: Int = 0,
  3.     extraBufferCapacity: Int = 0,
  4.     onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
  5. )

默认都为0，所以不会重复多次发送，这个就很好规避倒灌的问题。不会因为生命周期发生变化，内部出现多次订阅和消费的情况。

share流也是要在协程里的，emit协程体中，在emit之前不能进行订阅(collect)，否则会出现阻塞，导致事件无法往下走，后面的emit无法发出，可以在多个协程体中进行订阅。

  1.  
  2.         lifecycleScope.launch {
  3.  
  4.             //多次订阅也不会发送，只会优先发给第一个订阅的人
  5.             shareFloow.collect {  value ->
  6.                 textview.append("1collect="+value+"\n")
  7.             }
  8.             shareFloow.collect {  value ->
  9.                 textview.append("2collect="+value+"\n")
  10.             }
  11.  
  12.             shareFloow.collectIndexed { index, value ->
  13.                 textview.append("collectIndexed="+value+"，${index}\n")
  14.             }
  15.         }
  16.  
  17.         findViewById<View>(R.id.btn_click_share)?.setOnClickListener {
  18.             var count=0
  19.  
  20.             /**
  21.              * flow是一直保持的，只要订阅一次，就会发送一次，而且上次的记录一直保持。collect需要通过协程来处理
  22.              * */
  23.             lifecycleScope.launch {
  24.  
  25.  
  26.                 shareFloow.emit("我在测试ShareFlow1")
  27.                 shareFloow.emit("我在测试ShareFlow2")
  28.                 shareFloow.collect {  value ->
  29.                     textview.append("in--1--collect="+value+"\n")
  30.                 }
  31.                 shareFloow.collect {  value ->
  32.                     textview.append("in--2--collect="+value+"\n")
  33.                 }
  34.             }
  35.  
  36.         }

3.4Channel

        channel其实在很多地方都有，管道。Java中也有。这里也是可以通过管道channel来进行转发和订阅的。这个用法可以将管道转成flow，进行各种订阅

  1.        //后订阅这，不再接受到之前的事件
  2.         val channel= Channel<String>()
  3.         var count=0
  4.         findViewById<View>(R.id.btn_click_channel).setOnClickListener {
  5.  
  6.             lifecycleScope.launch{
  7.  
  8.                 channel.send( "count=1")
  9.                 channel.send( "count=2")
  10.  
  11.             }
  12.  
  13.  
  14.             }
  15.  
  16.         lifecycleScope.launch {
  17.  
  18.             //类似shareflow一直订阅，一直
  19. //            channel.consumeEach {
  20. //                textview.append("consumeEach=${it}\n")
  21. //            }
  22.             //只能订阅一次，且接受是第一次，后面不再接受
  23. //            channel.consume {
  24. //                textview.append("consume=${this.receive()}\n")
  25. //            }
  26.  
  27.  
  28.             //只消费一次，无法再继续接受，只接受第一次发送的事件
  29. //            channel.consumeAsFlow().produceIn(this).consume{
  30. //                textview.append("consumeAsFlow22=${receive()}\n")
  31. //            }
  32.             //会一直接受，和shareflow类似
  33.             channel.consumeAsFlow().collect{
  34.                 textview.append("consumeAsFlow11=${it}\n")
  35.             }
  36.  
  37.  
  38.         }

他的订阅类型很多

1.consumeEach：一直订阅，只要发送就会接受

2.consume：单次消费，订阅后，只接受第一次，后面再发送也不会接受

3.channel.consumeAsFlow()：转成热流

4.channel.consumeAsFlow().collect:热流的用法

四、总结

通过上面的分析，不管冷流还是热流以及管道，都有自己的特性。

冷热流以及管道可以互相转换

也可以转成LiveData

所以，大家在用的时候要注意，如果需要注意倒灌可以优先考虑ShareFlow，需要长期订阅用flow

注意：

但是sateFlow会出现倒灌情况，和LiveData一样，如何规避？其实可以用ShareFlow替代

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号