Kotlin知识体系之Flow

Kotlin知识体系

热流和冷流

StateFlow是如何比较是否需要更新呢？

private fun updateState(expectedState: Any?, newState: Any): Boolean {
        var curSequence = 0
        var curSlots: Array<StateFlowSlot?>? = this.slots // benign race, we will not use it
        synchronized(this) {
            val oldState = _state.value
            if (expectedState != null && oldState != expectedState) return false // CAS support
            if (oldState == newState) return true // Don't do anything if value is not changing, but CAS -> true
            _state.value = newState
            curSequence = sequence
            if (curSequence and 1 == 0) { // even sequence means quiescent state flow (no ongoing update)
                curSequence++ // make it odd
                sequence = curSequence
            } else {
                // update is already in process, notify it, and return
                sequence = curSequence + 2 // change sequence to notify, keep it odd
                return true // updated
            }
            curSlots = slots // read current reference to collectors under lock
        }
        /*
           Fire value updates outside of the lock to avoid deadlocks with unconfined coroutines.
           Loop until we're done firing all the changes. This is a sort of simple flat combining that
           ensures sequential firing of concurrent updates and avoids the storm of collector resumes
           when updates happen concurrently from many threads.
         */
        while (true) {
            // Benign race on element read from array
            curSlots?.forEach {
                it?.makePending()
            }
            // check if the value was updated again while we were updating the old one
            synchronized(this) {
                if (sequence == curSequence) { // nothing changed, we are done
                    sequence = curSequence + 1 // make sequence even again
                    return true // done, updated
                }
                // reread everything for the next loop under the lock
                curSequence = sequence
                curSlots = slots
            }
        }
    }

重点是这句if (oldState == newState) return true，所以是通过==来进行比较的

== 操作符

• 结构相等（内容相等）的比较，等同于调用 equals() 方法。
• 它用于比较两个对象的内容是否相等。
• 即使两个对象的内存地址不同，只要它们的内容相等，== 仍然会返回 true

=== 操作符

• 引用相等的比较，用来检查两个对象是否是同一个引用。
• 它比较的是两个对象在内存中的地址是否相同。
• 只有当两个变量引用的是同一个对象实例时，=== 才会返回 true。

所以我们可以理解为，当你在 Kotlin 中使用 == 时，编译器会将它转换为对 equals() 方法的调用。
一般来说，StateFlow我们会使用data class，在 Kotlin 中，数据类（data class）会自动为你生成 equals() 方法，它会根据所有的属性值来比较对象是否相同，默认的 equals()大致是如下这样实现的：

class Person(val name: String, val age: Int) {
    override fun equals(other: Any?): Boolean {
        if (this === other) return true  // 引用相等
        if (other !is Person) return false  // 类型检查
        // 比较内容相等
        return name == other.name && age == other.age
    }
    override fun hashCode(): Int {
        return name.hashCode() * 31 + age
    }
}

所以我们现在可以总结出来StateFlow的默认比较策略：

• 使用 == 比较内容
  
  • 比较地址
    地址相同，认为相等，不更新
  • 比较内容
    内容相同，认为相等，不更新
    
    • 嵌套data class
      如果存在嵌套的对象，那么默认也是使用 ==来比较

所以针对这个策略，我们想要保证StateFlow更新，就需要如下对策：

• 外层对象每次都需要new一个新对象
• 嵌套的对象也需要new一个对象
• 或者对象都重写equeals方法，去除===，不要比较地址

flowOn

• flowOn 只影响它前面的操作符
• flowOn 后面的操作符使用收集协程的上下文
• 每个 flowOn 都会创建一个新的协程上下文作用域

// dataSource
private val _appListFlow = MutableStateFlow<List<AppInfo>>(emptyList())
private val appListFlow: StateFlow<List<AppInfo>> = _appListFlow.asStateFlow()
override fun getAppList(): Flow<List<AppInfo>> {
    return appListFlow
}
// repository
fun getAppList(): Flow<List<AppInfo>> {
    return getAppList().map { appList ->
        val finalList = processAppList(appList)
        BLogger.i(TAG, "getAppList -> appList.size: ${finalList.size}, ${Thread.currentThread().name}")
        finalList
    }.flowOn(Dispatchers.IO)
}
// viewModel
mainScope.launch {
    appRepository.getAppList().collect {
        // 遍历打印
        var index = 1
        for (appInfo in it) {
            BLogger.i("${index++} -> $appInfo")
            BLogger.i("----------------------------------")
        }
    }
}

咱们简化一下：

// ✅ 正确用法：flowOn 影响前面的操作符
mainScope.launch {
    getAppList() // 看更新/发射值的协程调度器
        .map { ... } // 在子线程处理
        .flowOn(Dispatchers.IO)
        .collect { ... } // 在收集流的协程中指定调度器，所以在主线程收集
}

• 想要控制生产线程？
  在更新/发射值的协程中指定调度器，就是调用emit、tryEmit或者StateFlow赋值的时候在那个线程

// 在主线程调用
fun updateOnMainThread() {
    _appListFlow.tryEmit(newList) // 在主线程执行
}
// 在子线程调用
fun updateOnBackground() {
    thread {
        _appListFlow.tryEmit(newList) // 在子线程执行
    }
}
// 在协程中调用
viewModelScope.launch(Dispatchers.IO) {
    _appListFlow.tryEmit(newList) // 在IO调度器线程执行
}

• 想要控制消费线程？
  在收集流的协程中指定调度器

更复杂的一个例子：

mainScope.launch {
    getAppList()
        .map { ... } // 在 IO 线程
        .filter { ... } // 在 IO 线程
        .flowOn(Dispatchers.IO)
        .map { ... } // 在 Default 线程
        .flowOn(Dispatchers.Default) // 添加第二个 flowOn
        .map { ... } // 在主线程（收集线程）
        .collect { ... } // 在主线程（收集线程）
}

flowOn只影响它前面的操作符

参考

https://juejin.cn/post/7169843775240405022