Flow深入浅出系列之使用Kotlin Flow自动刷新Android数据的策略

Flow深入浅出系列之在ViewModels中使用Kotlin Flows
Flow深入浅出系列之更聪明的分享 Kotlin Flows
Flow深入浅出系列之使用Kotlin Flow自动刷新Android数据的策略

Flow深入浅出系列之使用Kotlin Flow自动刷新Android数据的策略

讨论在Android应用程序中使用Kotlin Flow高效加载数据的用法。它是第二部分“更智能的共享Kotlin Flows”的直接延续，因为它重用了相同的概念，涵盖了另一个用例：用户界面的自动定期刷新。

简单的定期刷新

当无法准确确定UI显示的数据集何时发生变化，或者当变化过于频繁时，一种常见的策略是在屏幕可见时定期以固定间隔重新加载数据。

实现这个的最简单方法之一是从一个无限循环中创建一个Flow，在每次发射之间调用delay()：

fun tickerFlow(period: Duration): Flow<Unit> = flow {while (true) {emit(Unit)    delay(period)}
}

这相当于RxJava中带有固定发射值（Unit）和初始延迟为0的Observable.interval()符。

然后，使用map()或mapLatest()操作符转换这个Flow，在每次定时器的“tick”上执行加载操作并返回结果：


tickerFlow(REFRESH_INTERVAL).map {repository.loadSomeData()}

注意这两个操作符之间的微妙差异：

使用map()，整个Flow将在单个协程中按顺序执行，这意味着delay()只会在加载操作完成后开始运行。因此，每个加载操作将延迟前一个加载操作所花费的时间加上固定间隔。
使用mapLatest()，主协程将收集tickerFlow()的上游值，同时创建一个子协程以并发执行加载操作并收集结果，而不会挂起主协程。这意味着delay()将立即在上一个tick之后开始运行，每个加载操作将严格按计划开始。这也意味着间隔必须比典型的加载时间长，因为它将充当超时：当tickerFlow()发射新值时，如果先前的加载操作未能及时完成，将被取消。然后，新的子协程取代之前的子协程执行下一个加载操作。

这个简单的实现将始终在Flow收集开始或重新开始时立即触发新的加载操作，而不考虑前一次运行。对于几秒钟的短时间间隔来说已经足够了，当加载操作被认为是廉价的时候。

使刷新更智能以利用缓存

然而，对于较长的刷新间隔和需要更多资源的加载操作（如执行API调用）来说，上述算法并不是很高效：当一个暂时隐藏的屏幕再次变为可见并且Flow收集重新开始时，我们希望避免重新加载仍然被认为是新鲜数据的那些数据的无谓工作，因为这些数据将在接下来的几分钟或几小时内仍然保持更新。

在第一篇文章中，我们看到StateFlow通常用于缓存Flow的最新值并在多个订阅者之间共享。但是StateFlow和SharedFlow有其局限性，因为它们无法根据Activity的生命周期简单地暂停和恢复工作：如果屏幕隐藏时停止了底层的Flow收集，当屏幕再次可见时，它总是需要从头开始重新启动，使得缓存在这种情况下毫无用处。

在第二篇文章中，我们研究了创建一个自定义Flow操作符的方法，旨在解决这些限制：flowWhileShared()。它允许只将底层Flow的上游部分与生命周期关联起来，以便在与一些过滤逻辑结合使用时，可以避免Flow下游的繁重工作。这样就可以充分利用StateFlow的缓存。

事实证明，我们可以使用相同的操作符来实现更智能的tickerFlow()版本，旨在与StateFlow结合使用。synchronizedTickerFlow()具有生命周期感知性，只会在父StateFlow至少有一个订阅者时发射值。使其更智能的是，它还记住了下一次发射的时间，在没有订阅者的暂停恢复后，在发出下一个值之前它将先等待达到那个时间点。

fun synchronizedTickerFlow(period: Duration,subscriptionCount: StateFlow<Int>,timeSource: TimeSource = ElapsedRealTimeSource
): Flow<Unit> {return flow {var nextEmissionTimeMark: TimeMark? = nullflow {nextEmissionTimeMark?.let { delay(-it.elapsedNow()) }while (true) {emit(Unit)nextEmissionTimeMark = timeSource.markNow() + perioddelay(period)}}.flowWhileShared(subscriptionCount, SharingStarted.WhileSubscribed()).collect(this)}
}

让我们详细分析代码。

你可能会注意到的第一件事是，计时逻辑是基于在Kotlin标准库1.9.0版本中稳定的新kotlin.time API。其中包括类Duration，TimeSource和TimeMark。

由函数返回的主要Flow收集了一个内部使用的辅助Flow的输出。主Flow的作用仅是封装状态（nextEmissionTimeMark）并将其局部化到每个收集中，就像所有Flow操作符一样。辅助Flow是根据父生命周期启动和停止的（当提供的subscriptionCount达到零时立即停止），这要归功于flowWhileShared()操作符。

注意：我们并没有向共享策略SharingStarted.WhileSubscribed()传递任何超时值，因为停止和重新启动辅助Flow的成本很低。

辅助流的主要逻辑与tickerFlow()几乎相同，唯一的区别是只有当达到nextEmissionTimeMark时才会发射第一个“tick”。TimeMark表示一点时间，nextEmissionTimeMark是下一个发射需要发生的最早时间点。

当主Flow收集开始时，nextEmissionTimeMark最初为null，并且第一个tick立即发射，没有延迟。然后，每次发射之后，下一个发射的未来时间点通过使用markNow()从TimeSource检索当前时间点并将期间Duration添加到其中来计算。

当辅助Flow在暂停后重新启动且nextEmissionTimeMark不为null时，调用TimeMark上的elapsedNow()来计算到达那个时间点所需等待的时间（Duration），并对结果进行否定，因为elapsedNow()实际上返回TimeMark和现在之间流逝的时间，如果TimeMark在未来，则为负值。请注意，使用负Duration调用delay()没有效果，将立即返回，因此我们不需要单独处理下一个发射时间已经到达并且现在已经过去的情况。

《IT Crowd》第一集的截图，Roy在电话中问道：对不起，你是来自过去的吗？

正确使用TimeSource的重要性

为了使这段代码正常工作，必须使用基于单调时钟（monotonic clock）而不是挂钟（wall clock）的TimeSource。单调时钟是一种始终向前移动且无法调整或重置的时钟。kotlin.time已经提供了TimeSource.Monotonic，它基于JVM和Android上的System.nanoTime()。虽然这个时钟对于JVM来说足够好，但对于Android应用程序可能会造成问题，因为当设备的CPU进入深度休眠时，它会停止运行，而这可能发生在屏幕关闭后。这意味着如果用户解锁刚刚经历了10分钟深度休眠的Android设备并返回到应用程序，数据刷新将会延迟10分钟。

在Android上，更适合此用途的时钟是SystemClock.elapsedRealtimeNanos()，它是具有纳秒精度的单调时钟，并包含设备在深度休眠模式下花费的时间。由于官方的Kotlin Android Jetpack库尚未提供基于此时钟的TimeSource，因此我们自己创建一个：

object ElapsedRealTimeSource : AbstractLongTimeSource(DurationUnit.NANOSECONDS) {override fun read(): Long = SystemClock.elapsedRealtimeNanos()override fun toString(): String = "TimeSource(SystemClock.elapsedRealtimeNanos())"
}

由于TimeSource作为参数传递给synchronizedTickerFlow()，因此可以轻松地替换实现，例如使用TestTimeSource进行测试。

将它们整合在一起

以下是如何将synchronizedTickerFlow()与前文中描述的stateFlow()工厂函数结合使用的示例。作为提醒，这个工厂函数允许StateFlow与提供数据的基础Flow共享其subscriptionCount。

@OptIn(ExperimentalCoroutinesApi::class)
val results: StateFlow<Result> = stateFlow(viewModelScope, Result.Empty) { subscriptionCount ->synchronizedTickerFlow(REFRESH_PERIOD, subscriptionCount).mapLatest {repository.loadSomeData()}
}

以下是此代码对UI状态更改的逐步反应：

当UI首次可见并开始收集结果StateFlow时，synchronizedTickerFlow()将开始定期发出新值，这将触发加载最新数据。此数据将缓存在StateFlow中，并与所有当前和未来的订阅者共享；
当UI变得不可见并停止收集StateFlow时，基础Flow仍然活动，但ticker将不会发出任何新值，因此不会加载新数据。我们节省了资源；
当UI再次可见并重新开始收集时，它将立即接收到StateFlow的缓存值。在基础Flow中，ticker将恢复。但首先，它将等待下一次发出的计划时间到达，然后再发出任何内容。这样，数据将根据其有效性保存在StateFlow缓存中，而不是无条件地被替换。我们现在节省了更多的资源。
高级用例：共享时间参考
有时，屏幕需要分别查询多个数据源，并定期更新。如果数据与时间有关，则所有源都需要使用相同的时间点作为参考，以确保它们的结果彼此一致。
例如：您希望定期加载过去10分钟和接下来10分钟的日程表，并确保结果不重叠。

这个时间参考点可以使用synchronizedTickerFlow()周期性地更新，并使用stateFlow()进行缓存和共享，与前面的示例完全相同：

private val timeReferenceFlow: Flow<Instant> = stateFlow(viewModelScope, null) { subscriptionCount ->synchronizedTickerFlow(REFRESH_PERIOD, subscriptionCount).map { Instant.now() }
}.filterNotNull()

不同之处在于，每个数据源将连接到相同的timeReferenceFlow实例，并使用flowWhileShared()和distinctUntilChanged()的组合来确保只有当时间参考发生变化时才更新数据。结果也将缓存在StateFlow中：

val results1: StateFlow<Result> = stateFlow(viewModelScope, Result.Empty) { subscriptionCount ->timeReferenceFlow.flowWhileShared(subscriptionCount, SharingStarted.WhileSubscribed()).distinctUntilChanged().map { timeReference: Instant ->repository.loadDataForTime(timeReference)}
}

使用这种模式使一个StateFlow依赖于另一个StateFlow，通过subscriptionCount将UI生命周期从一个StateFlow传播和聚合到下一个StateFlow。以下是它们对UI状态更改的反应方式：

当UI开始收集第一个结果StateFlow时，其subscriptionCount将从0更新为1，并且flowWhileShared()将开始收集上游的timeReferenceFlow；
作为StateFlow本身，timeReferenceFlow的subscriptionCount也会从0更新为1，并且synchronizedTickerFlow()将在前一个值过期后立即唤醒并开始发出新值；
当更多的结果StateFlow开始收集相同的timeReferenceFlow时，它们都会增加其subscriptionCount并立即接收到缓存的时间参考。计时器保持活动状态，并将每个新计算的时间参考分发给所有订阅者；
当所有连接到timeReferenceFlow的结果StateFlow停止被UI收集时，其subscriptionCount最终将达到零，并且synchronizedTickerFlow()将停止发出新值。
我们已经证明了timeReferenceFlow也具有生命周期感知性，即使它从未直接被UI收集。仅当它至少有一个活跃的订阅者并且前一个值已过期时，其值才会更新。这样可以在整个应用程序中最大限度地利用缓存，同时确保所有结果彼此一致。