一、使用场景1.1 LiveData 还是 StateFlowLiveData 问题StateFlow 解决粘性事件(重放)按下Button弹出Toast当配置改变例如屏幕旋转时页面会销毁后重建观察者将再次订阅LiveData此时会再次弹出Toast。一样存在粘性事件问题。事件应该使用SharedFlow它默认回放0额外缓存0。数据不防抖(判重)更新的值和当前值相同onChange()依然会再次调用。可以使用Transformations的distinctUntilChanger()、SingleLiveEvent解决。会判断更新的值与当前值是否相同相同则不更新。只处理最终值(丢失数据)在一个UI刷新周期内连续postValue()更新数据只会显示最后一次的内容。我们希望更新的每一个值事件都会刷新UI而LiveData会丢弃中间值只处理最终值最新状态。setValue不存在该情况一样只关注状态只持有1个最新值。可使用SharedFlow处理事件只在主线程处理更新值的函数都是在主线程回调也是在主线程。协程随意切换线程。生命周期感知只在界面活跃状态下Start和Resume状态接收通知非活跃状态更新UI无意义浪费资源。在UI销毁时Destroy状态自动取消订阅避免内存泄漏。StateFlow执行在协程中lifecycleScope会在UI销毁时结束协程。launchWhenX会在进入X状态前等待、进入后执行、离开后挂起如果内部订阅了数据流挂起只是停止消费数据没有取消协程是无法阻止被订阅的数据流继续活跃生产数据。repeatOnLifecycle会在离开X状态时取消协程再次进入X状态再重开协程即围绕X状态的进出多次重新执行代码。没有默认值创建实例未赋值被调用会异常。构造必须传入初始化值null安全符合页面必须有初始状态的逻辑。1.2 Suspend 还是 Flow除非界面非常简单否则不要用作UI状态流使用如获取数据时往 Flow 里面发送 Loading、Success(value)、Failed(value) 等状态然后在 UI 中收集并判断加载不同界面。界面状态应使用专门的 UiState通过可观察容器StateFlow、LiveData、MutableState暴露给 UI 订阅且优先使用挂起函数。Flow应该纯粹获取值正确的使用场景需要返回多个值应该是轮询如先返回缓存再返回最新数据、超时重试等连续加工等需求。SuspendFlow值的数量一次性异步调用单值。数据流多个值。场景一次性数据例如文章内容。①在Repository中合并多个数据源这些数据可能随时发生变化。②数据随时变化需要观察例如Room等。1.3 哪种数据流FlowStateFlowSharedFlowChannel类型冷流数据只能在创建对象的时候定义生产方式。热流数据可以后期发送到流中。数据的生产消费时才会生产数据就像爱奇艺点播视频。不消费也会生产数据就像广播电台播放内容。数据的接收完整每次消费收到的都是从头开始的完整数据。无订阅者会丢弃值只能接收后续数据。通信无订阅者会挂起等待对方的接收。最新只持有单个且最新的值。历史可以回放历史数据。关系独立多个订阅者彼此之间独立。共享多个订阅者同时接收收到的值相同。分配多个订阅者轮流接收收到的值不是同一个值。关闭流会自动关闭取消协程或数据生产完。构造创建的不会自动关闭转换的启动模式配置为WhileSubscribed会超时关闭。构造函数创建的不会自动关闭协程构建器创建的会跟随协程关闭。二、相互转换2.1 改造回调API2.1.1 Callback → SuspendsuspendCancellableCoroutine对于已有的调用了回调的函数可以改造成挂起函数onSuccess()中调用resume()返回数据onFailure()中调用resumeWithException()返回异常。suspendCoroutine适用于改造SAM回调。public suspend inline fun T suspendCancellableCoroutine(crossinline block: (CancellableContinuationT) - Unit): T//1.回调接口 interface NetCallbackT { fun onSuccess(t: T) fun onError(excpption: Exception) } //2.在Model中封装回调API联网获取数据 fun getData(callback: NetCallbackString) { Thread { Thread.sleep(1000) //模拟联网操作 callback.onSuccess(数据) callback.onError(Exception(错误)) }.run() } //3.以前在ViewModel中封装功能 fun show() { getData(object : NetCallbackString { override fun onSuccess(t: String) {...} override fun onError(excpption: Exception) {...} }) } //4.现在在ViewModel中封装功能 suspend fun showWithSuspend():String suspendCancellableCoroutine { cancellableContinuation - getData(object : NetCallbackString { override fun onSuccess(t: String) { cancellableContinuation.resume(t)} override fun onError(excpption: Exception) {cancellableContinuation.resumeWithException(excpption)} }) }2.1.2 Callback → FlowcallbackFlow底层使用的sendChannel默认容量64满了会挂起直到消费出空位为了避免生产被挂起可以配置为CONFLATED或者UNLIMITED。fun T callbackFlow(block: suspend ProducerScopeT.() - Unit): FlowTsend ()发送数据。offer ()允许在协程外提交。sendBlocking ()尝试用offer失败则用runBlocking{ send() }阻塞式提交。awaitClose ()Flow关闭时执行用来释放资源注销回调函数未调用报错 IllegalStateException。fun showWithFlow(): FlowInt callbackFlow { //1.创建Flow并发送值实现Callback接口 val callback object: NetCallback { override fun onNextValue(value: Int) { try { offer(num) } catch(t: Throwable) {...} } override fun onError(ecxeption: Throwable) { cancel(CancellationException(API发生错误, exception)) } override fun onCompleted() close() } //2.注册回调传参使用并对API进行配置操作 getData(callback) //3.取消协程并注销回调用来释放API资源 awaitClose {...} }2.2 数据转换到LiveData2.2.1 Suspend → LiveDataliveData{ }public fun T liveData(context: CoroutineContext EmptyCoroutineContext,//区块所执行的协程上下文默认Main.immediate。timeoutInMs: Long DEFAULT_TIMEOUT, //没有观察者后多少毫秒后取消区块默认5s。BuilderInference block: suspend LiveDataScopeT.() - Unit //区块在LiveData被观察的时候执行): LiveDataTemit ()提交新值emitSource ()订阅其它LiveData从中获取值。.latestValue获取最新提交的值class MyViewModel : ViewModel() { //直接调用挂起函数赋值 val num1: LiveDataInt liveData { emit(getData()) } //还可以指定线程单独写耗时操作 val num2: LiveDataInt liveData(Dispatchers.IO) { emit(5) delay(1000) emit(3) } //从另一个LivaData获取更新结果 val aa MutableLiveData(10) val bb liveData{ emitSource(aa) } suspend fun getData(): Int withContext(Dispatchers.IO) { 3 } }2.2.2 Flow → LiveDataasLiveDatapublic fun T FlowT.asLiveData(context: CoroutineContext EmptyCoroutineContext,//区块所执行的协程上下文默认Main.immediatetimeoutInMs: Long DEFAULT_TIMEOUT //没有观察者后多少毫秒后取消区块默认5s): LiveDataTclass MyViewModel : ViewModel() { val num1: LiveDataInt liveData { DataSource().getDataFlow.collect { emit(it) } } val num2: LiveDataInt DataSource().getDataFlow.asLiveData() //简写 } class DataSource { val getDataFlow: FlowInt flow { repeat(3) { emit(it) } } }2.3 流之间转换2.3.1 Channel → Flowpublic fun T ReceiveChannelT.receiveAsFlow(): FlowT流能被多次消费。public fun T ReceiveChannelT.consumeAsFlow(): FlowT流只能被消费一次。fun main(): Unit runBlocking { val channel ChannelInt() val receiveAsFlow channel.receiveAsFlow() val consumeAsFlow channel.consumeAsFlow() launch { receiveAsFlow.collect { print($it,) } } repeat(3) { channel.send(it) } channel.close() }2.3.2 Flow → Channelpublic fun T FlowT.produceIn(scope: CoroutineScope): ReceiveChannelTfun main(): Unit runBlocking { val receiveChannel (1..3).asFlow().produceIn(this) repeat(3) { print($i,) } }2.3.3 Flow → SharedFlowpublic fun T FlowT.shareIn(scope: CoroutineScope, //数据共享时所在的协程作用域started: SharingStarted, //启动策略replay: Int 0 //回放新订阅时得到几个之前已经发射过的旧值。): SharedFlowT2.3.4 Flow → StateFlowpublic fun T FlowT.stateIn(scope: CoroutineScope, //数据共享时所在的协程作用域started: SharingStarted, //启动策略initialValue: T //默认值): StateFlowTpublic suspend fun T FlowT.stateIn(scope: CoroutineScope): StateFlowT {val config configureSharing(1)val result CompletableDeferredStateFlowT()scope.launchSharingDeferred(config.context, config.upstream, result)return result.await()}挂起函数版本不用指定默认值会挂起直到产出第一个值。三、UI 生命周期Flow 是通用框架不关心是在什么 UI 框架中订阅不同UI框架专有的生命周期处理也无法感知应用生命周期处于前台后台不会自动停止收集必须在协程中收集还要确保切到主线程中更新UI因此需要特殊处理。通常需要在UI层收集数据流以便显示更新当UI进入后台不可见时数据流也会持续发送不停止除非手动取消协程。如使用了基于Channel收集的CoroutineScope.launch、Flow.launchIn、LifecycleCoroutineScope.launchWhenX或使用了带缓存操作符的Flowbuffer、conflate、flowIn、shareIn。收集单个Flow使用Flow.flowWithLifecyle( )收集多个流、热流使用Lifecyle.repeatOnLifecyle( )不搜集流使用LifecycleScope.launchWhenX( )3.1 Flow.flowWithLifecyle( )当只有一个Flow需要收集时使用该操作符内部基于Lifecyle.repeatOnLifecyle()实现的扩展函数。生命周期低于目标状态会取消上游不影响下游需要注意调用顺序。public fun T FlowT.flowWithLifecycle(lifecycle: Lifecycle,minActiveState: Lifecycle.State Lifecycle.State.STARTED): FlowT callbackFlow {lifecycle.repeatOnLifecycle(minActiveState) {thisflowWithLifecycle.collect {send(it)}}close()}3.2 Lifecyle.repeatOnLifecyle( )在生命周期到达目标状态时挂起调用它的协程。会在离开目标状态时取消子协程再次进入目标状态会重开子协程即围绕目标状态的进出多次重新执行代码。进入Destroy状态才会恢复调用它的协程。public suspend fun Lifecycle.repeatOnLifecycle(state: Lifecycle.State,//可选状态INITIALIZED、CREATED、STARTED、RESUMED、DESTROYEDblock: suspend CoroutineScope.() - Unit)lifecycleScope.launch { //会挂起之前的代码 repeatOnLifecycle(Lifecycle.State.STARTED) { //重复的工作 } //当进入onDestroy()状态协程会恢复执行后面的代码 }3.3LifecycleScope.launchWhenX( )会在进入X状态前等待、进入后执行、离开后挂起。如果内部订阅了数据流挂起只是停止消费数据没有取消协程无法阻止被订阅的数据流继续活跃生产数据由于会造成资源浪费已被废弃使用上面的 3.2 替代。public fun launchWhenCreated(block: suspend CoroutineScope.() - Unit): Jobpublic fun launchWhenResumed(block: suspend CoroutineScope.() - Unit): Jobpublic fun launchWhenStarted(block: suspend CoroutineScope.() - Unit): Job3.4 配置生产者MutableSharedFlowMutableStateFlow持有数据流实例的对象还在内存中它们就会保持生产者的活跃状态。控制字段subscriptionCount为0的时候内部的生产者就会停止。Flow.shareInFlow.stateIn对形参启动策略配置为WhileSubscriibed()会在没有活跃订阅者时停止内部生产者。而无论是配置成Eagerly还是Lazy只要使用的协程作用域还处于活跃状态内部生产者就会保持活跃。