Android Kotlin Coroutines：flow、callbackFlow、channelFlow、...其他流构造函数有什么区别

Question

我有代码应该使用流将 SharedPreferences 更改为可观察的存储，所以我有这样的代码

internal val onKeyValueChange: Flow<String> = channelFlow {
        val callback = SharedPreferences.OnSharedPreferenceChangeListener { _, key ->
            coroutineScope.launch {
                //send(key)
                offer(key)
            }
        }

        sharedPreferences.registerOnSharedPreferenceChangeListener(callback)

        awaitClose {
            sharedPreferences.unregisterOnSharedPreferenceChangeListener(callback)
        }
    }

或这个

internal val onKeyValueChange: Flow<String> = callbackFlow {
        val callback = SharedPreferences.OnSharedPreferenceChangeListener { _, key ->
            coroutineScope.launch {
                send(key)
                //offer(key)
            }
        }

        sharedPreferences.registerOnSharedPreferenceChangeListener(callback)

        awaitClose {
            sharedPreferences.unregisterOnSharedPreferenceChangeListener(callback)
        }
    }

然后我观察令牌、userId、companyId 的这个偏好，然后登录，但有一点很奇怪，因为我需要构建应用程序三次，比如更改令牌不会导致 tokenFlow 发出任何东西，然后第二次新的 userId 不会导致 userIdFlow 发出任何东西，然后在第三次登录后，我可以注销/登录并且它可以工作。注销时，我正在清除 prefs 令牌、userId、companyId 中的所有 3 个属性存储。

Answer 1

对于callbackFlow：

您不能在回调中使用emit()简单Flow（因为它是一个suspend函数）。因此callbackFlow，该offer()选项为您提供了一种同步方式。

例子：

fun observeData() = flow {
 myAwesomeInterface.addListener{ result ->
   emit(result) // NOT ALLOWED
 }
}

因此，协程为您提供了以下选项callbackFlow：

fun observeData() = callbackFlow {
 myAwesomeInterface.addListener{ result ->
   offer(result) // ALLOWED
 }
 awaitClose{ myAwesomeInterface.removeListener() }
}

对于channelFlow：

文档中Flow描述了它和基本的主要区别：

使用具有默认缓冲区大小的通道。在结果流上使用缓冲区运算符来指定用户定义的值并控制当数据生成速度快于消耗时发生的情况，即控制背压行为。

将offer()仍然代表着同样的事情。这只是一种同步方式（非suspending方式）emit()或send()

我建议您查看 Romans Elizarov博客以获取更多详细信息，尤其是这篇文章。

关于您的代码，因为callbackFlow您不需要协程启动：

coroutineScope.launch {
                send(key)
                //offer(key)
            }

只需使用 offer()