Kotlin协程之Flow使用(三) 本章前言 kotlin

本章前言

这篇文章是kotlin协程系列的时候扩展而来，如果对kotlin协程感兴趣的可以通过下面链接进行阅读、

Kotlin协程基础及原理系列

史上最详Android版kotlin协程入门进阶实战(一) -> kotlin协程的基础用法
史上最详Android版kotlin协程入门进阶实战(二) -> kotlin协程的关键知识点初步讲解
史上最详Android版kotlin协程入门进阶实战(三) -> kotlin协程的异常处理
史上最详Android版kotlin协程入门进阶实战(四) -> 使用kotlin协程开发Android的应用
史上最详Android版kotlin协程入门进阶实战(五) -> kotlin协程的网络请求封装
史上最详Android版kotlin协程入门进阶实战(六) -> 深入kotlin协程原理(一)
史上最详Android版kotlin协程入门进阶实战(七) -> 深入kotlin协程原理(二)
[史上最详Android版kotlin协程入门进阶实战(八) -> 深入kotlin协程原理(三)]
[史上最详Android版kotlin协程入门进阶实战(九) -> 深入kotlin协程原理(四)]

Flow系列

扩展系列

封装DataBinding让你少写万行代码
ViewModel的日常使用封装
笔者也只是一个普普通通的开发者，设计不一定合理，大家可以自行吸收文章精华，去糟粕。

kotlin协程之Flow使用(三)

上一章节我们了解了StatedFlow的相关使用，数据更新基本原理，以及如何避免StatedFlow使用的一些坑。本章节我们主要讲解SharedFlow的使用，以及在实际开发使用过程的选择问题。

`SharedFlow`的使用

我们在上面使用StateFlow的时候就了解到，StateFlow是继承自SharedFlow，是SharedFlow一个更佳具体实现，同时我们也可以把SharedFlow看作是StateFlow的可配置性极高的泛化数据流。

SharedFlow用来取代BroadcastChannel。SharedFlow不仅使用起来更简单、更快速，而且比BroadcastChannel的功能更丰富。但在需求需要的时候，仍然可以使用Channels API。

SharedFlow也是两种类型: SharedFlow和MutableSharedFlow。与上面功能类似。SharedFlow是只读的，如果需要对值进行修改,则需要使用MutableSharedFlow。

但是与StateFlow不同的是，SharedFlow是无法在创建的时候设置初始默认值的。同时SharedFlow在初始的时候有3个可选配置项。

kotlin复制代码public fun <T> MutableSharedFlow(
    replay: Int = 0,
    extraBufferCapacity: Int = 0,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
): MutableSharedFlow<T> {
    require(replay >= 0) { "replay cannot be negative, but was $replay" }
    require(extraBufferCapacity >= 0) { "extraBufferCapacity cannot be negative, but was $extraBufferCapacity" }
    require(replay > 0 || extraBufferCapacity > 0 || onBufferOverflow == BufferOverflow.SUSPEND) {
        "replay or extraBufferCapacity must be positive with non-default onBufferOverflow strategy $onBufferOverflow"
    }
    val bufferCapacity0 = replay + extraBufferCapacity
    val bufferCapacity = if (bufferCapacity0 < 0) Int.MAX_VALUE else bufferCapacity0 
    return SharedFlowImpl(replay, bufferCapacity, onBufferOverflow)
}

replay：重新发射给新的订阅者的值的数量，可以将旧的数据回播给新的订阅者。不能为负数，默认为0。
extraBufferCapacity：在replay基础上的缓冲池的数量，当有剩余缓冲区空间时，调用emit发射数据不会被挂起，同样的不能为负数，默认值为0。
onBufferOverflow：配置一个emit在缓冲区溢出时的触发操作。默认为BufferOverflow.SUSPEND，缓存溢出时挂起。另外还有BufferOverflow.DROP_OLDEST在溢出时删除缓冲区中最旧的值，将新值添加到缓冲区，不会进行挂起。BufferOverflow.DROP_LATEST在缓冲区溢出时删除当前添加到缓冲区的最新值来保持缓冲区内容不变，不会进行挂起。

通过上面可以看到，MutableSharedFlow创建以后，最终会返回一个SharedFlowImpl对象。我们先用SharedFlow实现上面StateFlow的例子：

kotlin复制代码class TestActivity : AppCompatActivity() {
    private val viewModel: TestFlowViewModel by viewModels()
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.test)
        lifecycleScope.launch {
            viewModel.state.collect {
                Log.d("carman", "state : $it")
            }
        }
        viewModel.download()
    }
}

class TestFlowViewModel : ViewModel() {
    private val _state: MutableSharedFlow<Int> = MutableSharedFlow()
    val state: SharedFlow<Int> get() = _state
    fun download() {
        for (state in 0..5) {
            viewModelScope.launch(Dispatchers.IO) {
                delay(100L * state)
                _state.emit(state)
            }
        }
    }
}

kotlin复制代码 D/carman: state : 0
 D/carman: state : 1
 D/carman: state : 2
 D/carman: state : 3
 D/carman: state : 4
 D/carman: state : 5

可以看到默认的参数这里使用跟StateFlow没什么区别。

我们现在修改一下创建MutableSharedFlow时候的参数replay,同时再新增一个收集操作：

kotlin复制代码class TestActivity : AppCompatActivity() {
    private val viewModel: TestFlowViewModel by viewModels()
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.test)
        lifecycleScope.launch {
            var index = 1
            launch {
                viewModel.state.collect {
                    Log.d("carman", "第一个 state : $it ")
                }
            }
           launch {
               delay(3000)
               viewModel.state.collect {
                   Log.d("carman", "第二个 state : $it")
               }
           }
        }
        viewModel.download()
    }
}

class TestFlowViewModel : ViewModel() {
    private val _state: MutableSharedFlow<Int> = MutableSharedFlow(2)
    val state: SharedFlow<Int> get() = _state
    fun download() {
        for (state in 0..5) {
            viewModelScope.launch(Dispatchers.IO) {
                delay(100L * state)
                _state.emit(state)
            }
        }
    }
}

我们这里在第二个collect中通过delay延时3秒，来确保第一个接收完成后第二个才开始进行数据收集，这里为了更加清晰，把日志打印时间也一并显示：

kotlin复制代码03:37:08.412 D/carman: 第一个 state : 0 
03:37:08.487 D/carman: 第一个 state : 1 
03:37:08.586 D/carman: 第一个 state : 2 
03:37:08.686 D/carman: 第一个 state : 3 
03:37:08.786 D/carman: 第一个 state : 4 
03:37:08.886 D/carman: 第一个 state : 5 
03:37:11.383 D/carman: 第二个 state : 4
03:37:11.383 D/carman: 第二个 state : 5

这时候我们就可以看到，在第一个collect接收完所有变化的数据以后，当我们再次启动一个新的collect时，第二个collect函数里面会接收到两次数据，而且是最新的两次数据4和5。这里是如何实现的呢。

那么我们就需要继续看MutableSharedFlow最终返回的SharedFlowImpl对象的源码实现：

kotlin复制代码private class SharedFlowImpl<T>(
      private val replay: Int,
      private val bufferCapacity: Int,
      private val onBufferOverflow: BufferOverflow
) : AbstractSharedFlow<SharedFlowSlot>(), MutableSharedFlow<T>, CancellableFlow<T>, FusibleFlow<T> {

      private var buffer: Array<Any?>? = null
      private var replayIndex = 0L
      private var minCollectorIndex = 0L
      private var bufferSize = 0
      private var queueSize = 0

      private val head: Long get() = minOf(minCollectorIndex, replayIndex)
      private val replaySize: Int get() = (head + bufferSize - replayIndex).toInt()
      private val totalSize: Int get() = bufferSize + queueSize
      private val bufferEndIndex: Long get() = head + bufferSize
      private val queueEndIndex: Long get() = head + bufferSize + queueSize

      override val replayCache: List<T>
            get() = synchronized(this) {
                  val replaySize = this.replaySize
                  if (replaySize == 0) return emptyList()
                  val result = ArrayList<T>(replaySize)
                  val buffer = buffer!!
                  @Suppress("UNCHECKED_CAST") 
                  for (i in 0 until replaySize) result += buffer.getBufferAt(replayIndex + i) as T
                  result
            }
       //...
}

我们这里先过滤掉一些实现方法，这里的数据更新实现还挺复杂的，我们主要看SharedFlowImpl中定义的一些属性，这里我们要分为三部分来理解：

用于存储状态的部分：

buffer：缓冲数组，创建和每次分配的大小总是2的幂。
replayIndex：从新收集器(订阅者)中获取值的最小索引。也就是重新发射给新的订阅者的值的数量的位置索引，会根据更新位置的变化，而变化。
minCollectorIndex：活动收集器的最小索引，如果没有则等于replayIndex
bufferSize：缓冲的数量
queueSize：排队的发射器数量

用于计算状态的部分：

head：头部索引，取得是replayIndex和minCollectorIndex中最小的值。
replaySize：重新发射给新的订阅者的值的数量大小，由创建的时候replay决定。
totalSize：总得数量，bufferSize和queueSize之和。
bufferEndIndex：缓冲池的尾部索引
queueEndIndex：发射器队列的尾部索引

获取缓存数据部分：

replayCache:缓存数据快照，集合的大小由创建的时候replay决定，也就是用于计算部分的replaySize变量。每个新订阅者优先从缓存快照中获取值，然后获得新的触发值。可以通过MutableSharedFlowd的resetReplayCache函数重置。

附加一个官方的缓冲区的逻辑结构解释图：

doc复制代码             buffered values
        /-----------------------\
                     replayCache      queued emitters
                     /----------/----------------------\
    +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
    |   | 1 | 2 | 3 | 4 | 5 | 6 | E | E | E | E | E | E |   |   |   |
    +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
          ^           ^           ^                      ^
          |           |           |                      |
         head         |      head + bufferSize     head + totalSize
          |           |           |
index of the slowest  |    index of the fastest
 possible collector   |     possible collector
          |           |
          |     replayIndex == new collector's index
          ---------------------- /
     range of possible minCollectorIndex

上面的案例中，我们只是观察接收到的数据是看不太大的变化。通过上面的一些变量知道，这个时候我们需要观察SharedFlow的replayCache属性：

kotlin复制代码class TestActivity : AppCompatActivity() {
    private val viewModel: TestFlowViewModel by viewModels()
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.test)
        lifecycleScope.launch {
            var index = 1
            viewModel.state.collect {
                Log.d("carman", "第${index++}次变化 state : $it   replayCache: ${viewModel.state.replayCache}")
            }
        }
        viewModel.download()
    }
}

class TestFlowViewModel : ViewModel() {
    private val _state: MutableSharedFlow<Int> = MutableSharedFlow(2)
    val state: SharedFlow<Int> get() = _state
    fun download() {
        for (state in 0..5) {
            viewModelScope.launch(Dispatchers.IO) {
                delay(200L * state)
                _state.emit(state)
            }
        }
    }
}

kotlin复制代码D/carman: 第一个 state : 0 replayCache: [0]
D/carman: 第一个 state : 1 replayCache: [0, 1]
D/carman: 第一个 state : 2 replayCache: [1, 2]
D/carman: 第一个 state : 3 replayCache: [2, 3]
D/carman: 第一个 state : 4 replayCache: [3, 4]
D/carman: 第一个 state : 5 replayCache: [4, 5]
D/carman: 第二个 state : 4 replayCache: [4, 5]
D/carman: 第二个 state : 5 replayCache: [4, 5]

现在我们就可以明显的看到数据变化的区别，当我们有数据变化的时候，SharedFlow会把新的数据存进buffer当中，每次有新的数据进来都会更新buffer。然后当有新订阅者时，优先从buffer中获取值，然后获得新的触发值。

而我们直接获取的replayCache只是获取的我们限定replay片段大小，通过replayIndex的索引位置获取指定大小的值得集合。

kotlin复制代码override val replayCache: List<T>
    get() = synchronized(this) {
        val replaySize = this.replaySize
        if (replaySize == 0) return emptyList()
        val result = ArrayList<T>(replaySize)
        val buffer = buffer!!
        @Suppress("UNCHECKED_CAST")
        for (i in 0 until replaySize) result += buffer.getBufferAt(replayIndex + i) as T
        result
    }

`ShareIn`转换

在我们日常使用中，我们都可以通过Flow的扩展方法ShareIn将一个Flow对象转换成SharedFlow，但是如果我们的对象不是Flow类型，我们可以通过asFlow先将它转换成Flow类型，比如：

kotlin复制代码class MainActivity : AppCompatActivity() {
    private lateinit var flow1: SharedFlow<Int>
    private lateinit var flow2: SharedFlow<Int>
    private lateinit var flow3: SharedFlow<Int>

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        lifecycleScope.launch {
            flow1 = mutableListOf(1, 2, 3, 4).asFlow().shareIn(this, SharingStarted.Eagerly, replay = 4)
            flow2 = arrayOf(5, 6, 7, 8).asFlow().shareIn(this, SharingStarted.Eagerly, replay = 4)
            flow3 = MutableStateFlow(9).shareIn(this, SharingStarted.Eagerly, replay = 1)

            launch {
                flow1.collect {
                    Log.d("carman", "flow1 : $it  replayCache: ${flow1.replayCache}")
                }
            }
            launch {
                flow2.collect {
                    Log.d("carman", "flow2 : $it replayCache: ${flow2.replayCache}")
                }
            }
            launch {
                flow3.collect {
                    Log.d("carman", "flow3 : $it replayCache: ${flow3.replayCache}")
                }
            }
        }
    }

kotlin复制代码D/carman: flow1 : 1  replayCache: [1, 2, 3, 4]
D/carman: flow1 : 2  replayCache: [1, 2, 3, 4]
D/carman: flow1 : 3  replayCache: [1, 2, 3, 4]
D/carman: flow1 : 4  replayCache: [1, 2, 3, 4]
D/carman: flow2 : 5 replayCache: [5, 6, 7, 8]
D/carman: flow2 : 6 replayCache: [5, 6, 7, 8]
D/carman: flow2 : 7 replayCache: [5, 6, 7, 8]
D/carman: flow2 : 8 replayCache: [5, 6, 7, 8]
D/carman: flow3 : 9 replayCache: [9]

可以看到，在案例中我们通过asFlow将数组，集合转换成Flow，然后再使用ShareIn将它转换成SharedFlow。

kotlin复制代码@FlowPreview
public fun <T> (() -> T).asFlow(): Flow<T> = flow {
    emit(invoke())
}

@FlowPreview
public fun <T> (suspend () -> T).asFlow(): Flow<T> = flow {
    emit(invoke())
}

public fun <T> Iterable<T>.asFlow(): Flow<T> = flow {
    forEach { value ->
        emit(value)
    }
}
 //... 
public fun IntArray.asFlow(): Flow<Int> = flow {
    forEach { value ->
        emit(value)
    }
}

上面展示的是部分asFlow的扩展函数，有兴趣了解全部的扩展函数，可以去源码下kotlinx.coroutines.flow包中去查看。

到此为止，我们关于Flow的文章就结束。

内存泄露问题

StateFlow、SharedFlow与LiveData 具有相似之处。两者都是可观察的数据容器类，并且在应用架构中使用时，两者都遵循相似模式。但请他们与 LiveData 的行为又有所不同

当 View 进入 TOPPED 状态时，LiveData.observe() 会自动取消注册使用方，而从StateFlow或任何其他数据流收集数据的操作并不会自动停止。即使View不可见，这些函数也会处理事件。此s时该行为可能会导致应用崩溃。为避免这种情况，需要使用repeatOnLifecycle API 。

kotlin复制代码class TestActivity : AppCompatActivity() {
    private val viewModel: TestFlowViewModel by viewModels()
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.test)
        lifecycleScope.launch {
            repeatOnLifecycle(Lifecycle.State.STARTED) {
                viewModel.state.collect {
                    Log.d("carman", "state : $it")
                }
            }
        }
        viewModel.download()
    }
}

我们从repeatOnLifecycle的源码实现可以看到，他们是通过观察对应组件对应的生命周期来防止内存泄露。

注意：repeatOnLifecycle API 仅在 androidx.lifecycle:lifecycle-runtime-ktx:2.4.0-alpha01 库及更高版本中提供。如果我们不使用此方法，那么我们需要把launch以后的Job对象保存起来，然后在相应的阶段cancel掉就可以了。

kotlin复制代码class TestActivity : AppCompatActivity() {
    private val viewModel: TestFlowViewModel by viewModels()
    private var job: Job?= null
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.test)
        job = lifecycleScope.launch {
            viewModel.state.collect {
                Log.d("carman", "state : $it")
            }
        }
    }

    override fun onDestroy() {
        super.onDestroy()
        job?.cancel()
    }
}

原创不易。如果您喜欢这篇文章，您可以动动小手点赞收藏。

关联文章
Kotlin协程基础及原理系列

史上最详Android版kotlin协程入门进阶实战(一) -> kotlin协程的基础用法
史上最详Android版kotlin协程入门进阶实战(二) -> kotlin协程的关键知识点初步讲解
史上最详Android版kotlin协程入门进阶实战(三) -> kotlin协程的异常处理
史上最详Android版kotlin协程入门进阶实战(四) -> 使用kotlin协程开发Android的应用
史上最详Android版kotlin协程入门进阶实战(五) -> kotlin协程的网络请求封装
史上最详Android版kotlin协程入门进阶实战(六) -> 深入kotlin协程原理(一)
史上最详Android版kotlin协程入门进阶实战(七) -> 深入kotlin协程原理(二)
[史上最详Android版kotlin协程入门进阶实战(八) -> 深入kotlin协程原理(三)]
[史上最详Android版kotlin协程入门进阶实战(九) -> 深入kotlin协程原理(四)]

Flow系列

扩展系列

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

本章前言

kotlin协程之Flow使用(三)

SharedFlow的使用

ShareIn转换

内存泄露问题

`kotlin`协程之`Flow`使用(三)

`SharedFlow`的使用

`ShareIn`转换