Flow 的理解与运用
Kotlin Flow 可以用于替换 Rxjava，也可以用于替换 LiveData，功能十分强大，它是 Kotlin 协程库提供的一部分功能，因此，如果我们项目中已经引用了 Kotlin 协程，则不需要额外引入 Flow 相关的依赖。
在协程中，挂起函数最多仅能返回一个值，而数据流 Flow 可按顺序发出多个值，例如，我们可以通过数据流从数据库中实时接收更新。数据流使用挂起函数通过异步方式生成和使用值，也就是说，数据流可安全地发出网络请求以生成下一个值，而不会阻塞主线程。
数据流 Flow 包含三个重要角色：
数据提供方：生成数据，并添加到数据流中
中介（可选）：可修改发送到数据流的值，或修正数据流本身
数据使用方：使用数据流中的值
创建数据流
flow 构建器函数会创建一个新数据流，然后我们可使用 emit 函数将新值发送到数据流中。
val latestNews: Flow<List<NewsData>> = flow {
while (true) {
val latestNews = newsApi.fetchLatestNews()
emit(latestNews)
delay(5000)
}
}
修改数据流
中介可以利用中间运算符在不使用值的情况下修改数据流，例如：
- filter : 对待操作的值进行过滤
- map ：对值进行加工后继续向后传递
- flatMapLatest：转换成一个新的流，需要返回一个转换后的新的流。（如果下个值来了，上一个值变换还没结束，上一个值的转换会被取消）
- onEach：接收到的每一个值
val news: Flow<List<NewsData>> = myRemoteDataSource.latestNews
// 先过滤列表数据大于3的数据（大于3才能通过）
.filter {
it.size >= 3
}
// 对结果进行加工后继续向后传递
.map { list ->
// 调用 list.filter 进一步筛选出 id==1 的新闻
list.filter { it.id == 1 }
}
// 转换成一个新的流，需要返回一个转换后的新的流。（如果下个值来了，上一个值变换还没结束，上一个值的转换会被取消）
.flatMapLatest {
flow {
emit(it)
}
}
.onEach {
// todo 获取到筛选后 新闻列表（结果） 数据
}
收集数据流
只有在 收集数据流时 才会触发 数据提供方 刷新最新数据。除非使用其他中间运算符指定流，否则数据流始终为冷流并延迟执行。
fun getNewsData() {
viewModelScope.launch(Dispatchers.Main) {
remoteRepository.news
.catch {
// todo 收集异常
}
.collect {
// 收到到的数据
}
}
}
数据流收集可能会由于以下原因而停止：
收集数据的协程被取消，此操作也会让 数据提供方 停止活动。
数据提供方完成发出数据项。在这种情况下，数据流将关闭，调用 collect 的协程则继续执行。
捕获异常
如需处理异常，可以使用 catch 运算符，如：
fun getNewsData() {
viewModelScope.launch {
remoteRepository.news
.catch {
// todo 在这里收集异常
}
.collect {
newsData.value = it
}
}
}
另外，catch 还可执行 emit 操作，向数据流发出新的数据项，例如，如果我们在 上游 发现了异常，我们可以继续调用 emit 函数发送新的数据（或者之前缓存的数据），如：
class MyRemoteRepository @Inject constructor(
private val myRemoteDataSource: MyRemoteDataSource,
) {
// 返回 id 等于 1 的新闻
val news: Flow<List<NewsData>> = myRemoteDataSource.latestNews
.map { news ->
// 筛选出 id==1 的新闻
news.filter { it.id == 1 }
}
.onEach {
// todo 获取到筛选后 新闻列表（结果） 数据
}
.catch {
// 如果在 上游 收集到异常，我们可以继续调用 emit 函数发送新的数据（或者之前缓存的数据）
// 例如：emit(lastCachedNews())
}
}
协程作用域切换
默认情况下，flow 上游数据提供方 会基于 下游数据收集方 的协程 CoroutineContext 执行，也就是说，默认情况，下游和上游会运行在同一个协程作用域中。
并且，它无法从不同协程作用域对值执行 emit 操作。
如果需要更改数据流的的协程作用域，可以使用中间运算符 flowOn 运算符。
flowOn 会更改上游数据流的 CoroutineContext，但不会影响到下游数据流的作用域。
如果有多个 flowOn 运算符，每个运算符都会更改当前位置的上游数据流。
// 上游数据流代码,上游数据流将会在 Dispatchers.IO 作用域上执行：
class MyRemoteRepository @Inject constructor(
private val myRemoteDataSource: MyRemoteDataSource,
) {
val news: Flow<List<NewsData>> = myRemoteDataSource.latestNews
.flowOn(Dispatchers.IO)
.catch { }
}
// 下游数据流代码，下游数据流将会在 Dispatchers.Main 作用域上执行
fun getNewsData() {
viewModelScope.launch(Dispatchers.Main) {
remoteRepository.news.collect { }
}
}
Flow Demo 演示
下面的代码将演示通过 Flow 不断获取最新的新闻列表。
步骤一：创建 News 类，定义新闻格式
data class NewsData(var id: Int, var content: String)
1
步骤二：创建 NewsApi 接口，用于请求最新的新闻列表
interface NewsApi {
/**
* 请求最新的新闻列表
*/
suspend fun fetchLatestNews(): List<NewsData>
companion object {
fun create(): NewsApi {
return NewsApiImpl()
}
}
}
步骤三：创建 DataSource 类，内部有一个 latestNews 变量，作为 Flow 的上游数据提供者，每隔 5 秒，通过 newsApi 请求新闻数据，并调用 emit 方法将新闻数据发出
class MyRemoteDataSource @Inject constructor(
private val newsApi: NewsApi,
) {
val latestNews: Flow<List<NewsData>> = flow {
while (true) {
val latestNews = newsApi.fetchLatestNews()
emit(latestNews)
delay(5000)
}
}
}
步骤四：创建 RemoteRepository 类，内部有一个 news 变量，作为 Flow 的数据中间处理者，筛选数据，切换上游作用域，收集上游异常等都可以在这里处理。
class MyRemoteRepository @Inject constructor(
private val myRemoteDataSource: MyRemoteDataSource,
) {
// 返回 id 等于 1 的新闻
val news: Flow<List<NewsData>> = myRemoteDataSource.latestNews
.map { news ->
// 筛选出 id==1 的新闻
news.filter { it.id == 1 }
}
.onEach {
// todo 获取到筛选后 新闻列表（结果） 数据
}
.flowOn(Dispatchers.IO) // 上游数据流将会在 Dispatchers.IO 作用域上执行
.catch {
// 如果在 上游 收集到异常，我们可以继续调用 emit 函数发送新的数据（或者之前缓存的数据）
// 例如：emit(lastCachedNews())
}
}
步骤五：创建 ViewModel 类，定义了一个成员方法 getNewsData() ，作为 Flow 的下游数据接收者，另外，还定义了一个 LiveData 变量，监听最新的新闻数据。
@HiltViewModel
class MainViewModel @Inject constructor(
private val remoteRepository: MyRemoteRepository,
) : ViewModel() {
val newsData = MutableLiveData<List<NewsData>>()
fun getNewsData() {
viewModelScope.launch(Dispatchers.Main) {
remoteRepository.news
.catch {
// todo 收集异常
}
.collect {
newsData.value = it
}
}
}
}
步骤六：编写 MainActivity 代码，接受最新的新闻数据并打印出来
@AndroidEntryPoint
class MainActivity : AppCompatActivity() {
private val mMainViewModel: MainViewModel by viewModels()
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
setContentView(R.layout.activity_main)
initObserve()
}
private fun initObserve() {
mMainViewModel.newsData.observe(this) {
println("newsData=${Gson().toJson(it)}")
}
}
fun click(view: View) {
mMainViewModel.getNewsData()
}
}
步骤七：创建 NewsApi 的实现类
class NewsApiImpl : NewsApi {
override suspend fun fetchLatestNews(): List<NewsData> {
val list = ArrayList<NewsData>()
list.add(NewsData(1, "news 1"))
list.add(NewsData(2, "news 2"))
list.add(NewsData(3, "news 3"))
return list
}
}
步骤八：编写 依赖注入类（di）
@InstallIn(SingletonComponent::class)
@Module
class MainModule {
@Singleton
@Provides
fun provideAppDatabase(@ApplicationContext context: Context): NewsApi {
return NewsApi.create()
}
}
// 另外，别忘了在 Application 中加上 @HiltAndroidApp 注解
@HiltAndroidApp
class MainApplication : Application()
步骤九：由于 demo 使用到了 Hilt ，因此我们需要加上如下依赖：
// Project-build.gradle
buildscript {
ext {
hiltVersion = '2.41'
}
dependencies {
classpath "com.google.dagger:hilt-android-gradle-plugin:$hiltVersion"
}
}
// app-build.gradle
plugins {
id 'kotlin-kapt'
id 'dagger.hilt.android.plugin'
}
dependencies {
kapt "com.google.dagger:hilt-android-compiler:$rootProject.hiltVersion"
implementation "com.google.dagger:hilt-android:$rootProject.hiltVersion"
}
步骤十：为了能在 activity 或 fragment 中使用 by viewModels() api，我们还需额外引入以下 依赖：
// app-build.gradle
dependencies {
implementation "androidx.activity:activity-ktx:1.4.0"
implementation "androidx.fragment:fragment-ktx:1.4.1"
}
StateFlw
StateFlow 是一个状态容器式的可观察数据流，可以向其收集器发出当前状态更新和新状态更新。还可通过其 value 属性读取当前状态值。
StateFlow 非常适合需要让可变状态保持可观察的类。
Flow 是冷数据流，而 StateFlow 是热数据流，热流有如下特性：
- 调用 StateFlow.collect 收集数据不会触发任何 数据提供方（上游） 的代码
- 上游数据流 如果已经处于活跃（发送）状态，即使没有任何地方调用 StateFlow.collect ，上游流仍会持续活跃（没有 Gc Root 引用自然也会被回收）
- 它允许被多个观察者共用 （因此是共享的数据流）
当一个 新的数据接收方 开始从数据流中 collect 数据时，它将接收到信息流中的最近一个状态及任何后续状态。
注意：如果 StateFlow.value 接收的新数据和前一个旧数据一样时，下游并不会接收到数据的更新通知。
StateFlow 和 LiveData
StateFlow 和 LiveData 具有相似之处，两者都是可观察的数据容器类。
但也存在不同之处：
- StateFlow 需要将初始状态传递给构造函数，而 LiveData 不需要
- 当 View 进入 STOPPED 状态时，LiveData.observe() 会自动取消注册使用方，但是从 StateFlow 或任何其他数据流收集数据的操作并不会自动停止，如需实现与 LiveData 相同的行为，则需从 Lifecycle.repeatOnLifecycle 块中收集数据流
StateFlow 的简单用法如下：
// ViewModel: (_uiState.value更新的地方属于上游)
@HiltViewModel
class MainViewModel @Inject constructor(
private val remoteRepository: MyRemoteRepository,
) : ViewModel() {
// 定义一个私有的 MutableStateFlow 变量（可变）
private val _uiState = MutableStateFlow(LatestNewsUiState.Success(emptyList()))
// UI 从此 StateFlow 收集以获取其状态更新
val uiState: StateFlow<LatestNewsUiState> = _uiState
fun getNewsData() {
viewModelScope.launch {
remoteRepository.news.collect {
// 接受到最新的新闻列表数据后，将数据赋值给 StateFlow 的 value
_uiState.value = LatestNewsUiState.Success(it)
}
}
}
}
// Activity: (mMainViewModel.uiState.collect的地方属于下游)
@AndroidEntryPoint
class MainActivity : AppCompatActivity() {
private val mMainViewModel: MainViewModel by viewModels()
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
setContentView(R.layout.activity_main)
initObserve()
}
private fun initObserve() {
mMainViewModel.viewModelScope.launch {
repeatOnLifecycle(Lifecycle.State.STARTED) {
mMainViewModel.uiState.collect {
// 注意：如果 StateFlow.value 接收的新数据和前一个旧数据一样时，下游并不会接收到数据的更新通知
when (it) {
is LatestNewsUiState.Success -> {
println("获取新闻成功，news=${Gson().toJson(it.news)}")
}
is LatestNewsUiState.Error -> {
println("获取新闻失败，error=${Gson().toJson(it.exception)}")
}
}
}
}
}
}
fun click(view: View) {
mMainViewModel.getNewsData()
}
}
如下代码中，负责更新 MutableStateFlow 的类是 数据提供方（上游） ，从 StateFlow.collect 的类是 数据使用方（下游）。
另外，repeatOnLifecycle 能使界面处于活跃状态下才会更新界面，要使用该 api，还需引入以下依赖：
// Project-build.gradle
buildscript {
ext {
lifecycleVersion = '2.4.1'
}
}
// app-build.gradle
dependencies {
implementation "androidx.lifecycle:lifecycle-runtime-ktx:$rootProject.lifecycleVersion"
}
Flow 转为 StateFlow
如需将任何数据流转换为 StateFlow ，可以使用 stateIn 中间运算符。
stateIn 有两个重载函数，一般我们用第二个：
// 函数1：
public suspend fun <T> Flow<T>.stateIn(scope: CoroutineScope): StateFlow<T>
// 函数2：
public fun <T> Flow<T>.stateIn(
scope: CoroutineScope,
started: SharingStarted,
initialValue: T
): StateFlow<T>
其中函数1是一个挂起函数，且仅需要传一个 scope 参数既可，函数2是非挂起函数，需要传递三个参数，三个参数的含义如下：
1、scope ：共享流开始时所在的协程作用域范围
2、started ：控制共享的开始和结束的策略
3、initialValue： 流的初始值
而 started 有三种取值可选：
- SharingStarted.Eagerly ：立即启动上游数据流，且在 scope 指定的作用域被结束时终止上游流
- SharingStarted.Lazily ：在第一个订阅者出现后开始启动上游数据流，且在 scope 指定的作用域被结束时终止上游流
- SharingStarted.WhileSubscribed(stopTimeoutMillis) ：在第一个订阅者出现后开始启动上游数据流，没有下游收集的情况下，指定时间后（默认是0）会取消上游数据流
stateIn 使用如下：
// ViewModel:
@HiltViewModel
class MainViewModel @Inject constructor(
private val remoteRepository: MyRemoteRepository,
) : ViewModel() {
// 将 news(Flow冷流) 转为 StateFlow 热流
val mStateFlow: StateFlow<List<NewsData>> = remoteRepository.news
.stateIn(
scope = viewModelScope,
started = WhileSubscribed(5000),
initialValue = emptyList()
)
}
// Activity:
@AndroidEntryPoint
class MainActivity : AppCompatActivity() {
private val mMainViewModel: MainViewModel by viewModels()
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
setContentView(R.layout.activity_main)
}
// 点击开始收集上游数据
fun click(view: View) {
mMainViewModel.viewModelScope.launch {
repeatOnLifecycle(Lifecycle.State.STARTED) {
mMainViewModel.mStateFlow.collect {
println("收集到数据，it=${Gson().toJson(it)}")
}
}
}
}
}
例子中的 ViewModel，定义了一个变量，将 Flow 冷流通过 stateIn 操作符转为 StateFlow ，该 StateFlow 在第一个订阅者出现后开始启动上游数据流，没有下游收集的情况下，会在 5秒 后取消上游数据流，另外，initialValue 初始值设置为 一个空列表 。在 MainActivity 的代码中，点击按钮将会触发 上游流开始发送数据，同时下游流也开始接收数据。
WhileSubscribed 传入了 5000 ，是为了实现等待5 秒后仍然没有订阅者存在就终止协程的功能，这个方法有以下功能：
应用转至后台运行后，5 秒钟后所有来自其他层的数据更新会停止，这样可以节省电量
在屏幕旋转时，因为重新订阅的时间在5s内，因此上游流不会中止
SharedFlow
SharedFlow 配置更为灵活，支持配置replay，缓冲区大小等，StateFlow是SharedFlow的特化版本，replay固定为1，缓冲区大小默认为0
我们可使用 shareIn 函数会返回一个热数据流 SharedFlow， SharedFlow 会 向从其所有的 数据接收方（下游） 发出数据。
我们先看下 ShareFlow 的构造函数：
public fun <T> MutableSharedFlow(
replay: Int = 0,
extraBufferCapacity: Int = 0,
onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
): MutableSharedFlow<T>
其主要有3个参数：
- replay：有新的订阅者Collect时，发送几个已经发送过的数据给它，默认为0
- extraBufferCapacity：除了 replay，SharedFlow 还缓存多少数据，默认为0
- onBufferOverflow：表示缓存策略，即缓冲区满了之后 ShareFlow 如何处理，默认为挂起
StateFolw 和 SharedFlow 的区别
StateFolw 和 SharedFlow 都属于热流。
StateFlow 本质上是一个 replay 为 1，且没有缓冲区的 SharedFlow，因此第一次订阅时会先获得默认值。
StateFlow 仅在值已更新，并且值发生了变化时才会返回，也就是说如果更新后的值没有变化，Collect 方法不会回调，但是 ShareFlow 是会回调的。
下面举个简单的使用 SharedFlow 的例子：
// ViewModel:
@HiltViewModel
class MainViewModel @Inject constructor(
private val remoteRepository: MyRemoteRepository,
) : ViewModel() {
// 定义一个私有的 MutableSharedFlow 变量（可变），当有新的订阅者时，会先发送1个之前发送过的数据给订阅者
private val mMutableSharedFlow = MutableSharedFlow<List<NewsData>>(replay = 1)
// 不可变的 shareFlow
val shareFlow: SharedFlow<List<NewsData>> = mMutableSharedFlow
fun getNewsData() {
viewModelScope.launch {
remoteRepository.news.collect {
mMutableSharedFlow.emit(it)
}
}
}
}
// Activity:
@AndroidEntryPoint
class MainActivity : AppCompatActivity() {
private val mMainViewModel: MainViewModel by viewModels()
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
setContentView(R.layout.activity_main)
}
// ShareFlow 开始发送数据
fun click(view: View) {
mMainViewModel.getNewsData()
}
// 收集 ShareFlow 发送的数据
fun click2(view: View) {
mMainViewModel.viewModelScope.launch {
repeatOnLifecycle(Lifecycle.State.STARTED) {
mMainViewModel.shareFlow.collect {
println("获取到的新闻，news=${Gson().toJson(it)}")
}
}
}
}
}
该例子中，ViewModel 中定义了一个 ShareFlow，并将其 replay 参数设置为 1 ，即当有新的订阅者时，会先发送 1 个之前发送过的数据给订阅者。在Activity 中，有两个按钮，按钮 1 触发 ShareFlow 上游开始发送数据， 按钮 2 触发 下游收集数据，按钮 2 按下后，下游会先收集到 1 个之前发送过的数据。
Flow 转为 SharedFlow
如需将任何数据流转换为 SharedFlow ，可以使用 ShareIn 中间运算符：
public fun <T> Flow<T>.shareIn(
scope: CoroutineScope,
started: SharingStarted,
replay: Int = 0
): SharedFlow<T>
ShareIn 函数有三个参数：
- scope：共享流开始时所在的协程作用域范围
- started：控制共享的开始和结束的策略
- replay：有新的订阅者Collect时，发送几个已经发送过的数据给它，默认为0
举个简单的例子（跟 StateIn 的例子很像）:
// ViewModel:
@HiltViewModel
class MainViewModel @Inject constructor(
private val remoteRepository: MyRemoteRepository,
) : ViewModel() {
// 将 news(Flow冷流) 转为 SharedFlow 热流
val mSharedFlow: SharedFlow<List<NewsData>> = remoteRepository.news
.shareIn(
scope = viewModelScope,
started = WhileSubscribed(5000),
replay = 1
)
}
// Activity:
@AndroidEntryPoint
class MainActivity : AppCompatActivity() {
private val mMainViewModel: MainViewModel by viewModels()
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
setContentView(R.layout.activity_main)
}
fun click(view: View) {
// 点击开始收集上游数据
mMainViewModel.viewModelScope.launch {
repeatOnLifecycle(Lifecycle.State.STARTED) {
mMainViewModel.mSharedFlow.collect {
println("收集到的新闻数据，news=${Gson().toJson(it)}")
}
}
}
}
}
例子中的 ViewModel，定义了一个变量，将 Flow 冷流通过 shareIn 操作符转为 SharedFlow ，该 SharedFlow 在第一个订阅者出现后开始启动上游数据流，没有下游收集的情况下，会在 5秒 后取消上游数据流，另外，replay设置为 1 ，当有下游收集者时，会将之前发过的最近一个值发给下游收集者。在 MainActivity 的代码中，点击按钮将会触发 上游流开始发送数据，同时下游流也开始接收数据。
