当前位置：首页 > news >正文

Kotlin 协程 - 在Android中的使用

news 2026/4/20 12:29:01

一、使用场景

1.1 LiveData 还是 StateFlow

LiveData 问题	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应该纯粹获取值，正确的使用场景（需要返回多个值）应该是轮询（如先返回缓存再返回最新数据）、超时重试等连续加工等需求。

Suspend

Flow

值的数量

一次性异步调用（单值）。

数据流（多个值）。

场景

一次性数据，例如文章内容。

①在Repository中合并多个数据源，这些数据可能随时发生变化。

②数据随时变化需要观察，例如Room等。

1.3 哪种数据流

Flow	StateFlow	SharedFlow	Channel
类型	冷流：数据只能在创建对象的时候定义生产方式。	热流：数据可以后期发送到流中。
数据的生产	消费时才会生产数据（就像爱奇艺点播视频）。	不消费也会生产数据（就像广播电台播放内容）。
数据的接收	完整：每次消费收到的都是从头开始的完整数据。	无订阅者会丢弃值，只能接收后续数据。		通信：无订阅者会挂起，等待对方的接收。
		最新：只持有单个且最新的值。	历史：可以回放历史数据。
关系	独立：多个订阅者彼此之间独立。	共享：多个订阅者同时接收，收到的值相同。		分配：多个订阅者轮流接收，收到的值不是同一个值。
关闭	流会自动关闭（取消协程或数据生产完）。	构造创建的不会自动关闭，转换的启动模式配置为WhileSubscribed会超时关闭。		构造函数创建的不会自动关闭，协程构建器创建的会跟随协程关闭。

二、相互转换

2.1 改造回调API

2.1.1 Callback → Suspend（suspendCancellableCoroutine）

对于已有的调用了回调的函数可以改造成挂起函数，onSuccess()中调用resume()返回数据，onFailure()中调用resumeWithException()返回异常。suspendCoroutine适用于改造SAM回调。

public suspend inline fun <T> suspendCancellableCoroutine(
crossinline block: (CancellableContinuation<T>) -> Unit
): T

//1.回调接口 interface NetCallback<T> { fun onSuccess(t: T) fun onError(excpption: Exception) } //2.在Model中封装回调API（联网获取数据） fun getData(callback: NetCallback<String>) { Thread { Thread.sleep(1000) //模拟联网操作 callback.onSuccess("数据") callback.onError(Exception("错误")) }.run() } //3.以前在ViewModel中封装功能 fun show() { getData(object : NetCallback<String> { override fun onSuccess(t: String) {...} override fun onError(excpption: Exception) {...} }) } //4.现在在ViewModel中封装功能 suspend fun showWithSuspend():String = suspendCancellableCoroutine { cancellableContinuation -> getData(object : NetCallback<String> { override fun onSuccess(t: String) { cancellableContinuation.resume(t)} override fun onError(excpption: Exception) {cancellableContinuation.resumeWithException(excpption)} }) }

2.1.2 Callback → Flow（callbackFlow）

底层使用的sendChannel，默认容量64满了会挂起直到消费出空位，为了避免生产被挂起可以配置为CONFLATED或者UNLIMITED。

fun <T> callbackFlow(block: suspend ProducerScope<T>.() -> Unit): Flow<T>

send ()	发送数据。
offer ()	允许在协程外提交。
sendBlocking ()	尝试用offer，失败则用runBlocking{ send() }阻塞式提交。
awaitClose ()	Flow关闭时执行，用来释放资源（注销回调函数），未调用报错 IllegalStateException。

fun showWithFlow(): Flow<Int> = 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 数据转换到LiveData

2.2.1 Suspend → LiveData（liveData{ }）

public fun <T> liveData(
context: CoroutineContext = EmptyCoroutineContext,//区块所执行的协程上下文，默认+Main.immediate。
timeoutInMs: Long = DEFAULT_TIMEOUT, //没有观察者后，多少毫秒后取消区块，默认5s。
@BuilderInference block: suspend LiveDataScope<T>.() -> Unit //区块在LiveData被观察的时候执行
): LiveData<T>

emit ()	提交新值
emitSource ()	订阅其它LiveData，从中获取值。
.latestValue	获取最新提交的值

class MyViewModel : ViewModel() { //直接调用挂起函数赋值 val num1: LiveData<Int> = liveData { emit(getData()) } //还可以指定线程，单独写耗时操作 val num2: LiveData<Int> = 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 → LiveData（asLiveData）

public fun <T> Flow<T>.asLiveData(
context: CoroutineContext = EmptyCoroutineContext,//区块所执行的协程上下文，默认+Main.immediate
timeoutInMs: Long = DEFAULT_TIMEOUT //没有观察者后，多少毫秒后取消区块，默认5s
): LiveData<T>

class MyViewModel : ViewModel() { val num1: LiveData<Int> = liveData { DataSource().getDataFlow.collect { emit(it) } } val num2: LiveData<Int> = DataSource().getDataFlow.asLiveData() //简写 } class DataSource { val getDataFlow: Flow<Int> = flow { repeat(3) { emit(it) } } }

2.3 流之间转换

2.3.1 Channel → Flow

public fun <T> ReceiveChannel<T>.receiveAsFlow(): Flow<T>

流能被多次消费。

public fun <T> ReceiveChannel<T>.consumeAsFlow(): Flow<T>

流只能被消费一次。

fun main(): Unit = runBlocking { val channel = Channel<Int>() val receiveAsFlow = channel.receiveAsFlow() val consumeAsFlow = channel.consumeAsFlow() launch { receiveAsFlow.collect { print("$it,") } } repeat(3) { channel.send(it) } channel.close() }

2.3.2 Flow → Channel

public fun <T> Flow<T>.produceIn(
scope: CoroutineScope
): ReceiveChannel<T>

fun main(): Unit = runBlocking { val receiveChannel = (1..3).asFlow().produceIn(this) repeat(3) { print("$i,") } }

2.3.3 Flow → SharedFlow

public fun <T> Flow<T>.shareIn(
scope: CoroutineScope, //数据共享时所在的协程作用域
started: SharingStarted, //启动策略
replay: Int = 0 //回放，新订阅时得到几个之前已经发射过的旧值。
): SharedFlow<T>

2.3.4 Flow → StateFlow

public fun <T> Flow<T>.stateIn(
scope: CoroutineScope, //数据共享时所在的协程作用域
started: SharingStarted, //启动策略
initialValue: T //默认值
): StateFlow<T>

public suspend fun <T> Flow<T>.stateIn(scope: CoroutineScope): StateFlow<T> {
val config = configureSharing(1)
val result = CompletableDeferred<StateFlow<T>>()
scope.launchSharingDeferred(config.context, config.upstream, result)
return result.await()
}

挂起函数版本，不用指定默认值，会挂起直到产出第一个值。

三、UI 生命周期

Flow 是通用框架，不关心是在什么 UI 框架中订阅（不同UI框架专有的生命周期处理），也无法感知应用生命周期处于前台后台（不会自动停止收集），必须在协程中收集还要确保切到主线程中更新UI，因此需要特殊处理。
通常需要在UI层收集数据流以便显示更新，当UI进入后台不可见时，数据流也会持续发送不停止（除非手动取消协程）。如使用了基于Channel收集的：CoroutineScope.launch、Flow.launchIn、LifecycleCoroutineScope.launchWhenX，或使用了带缓存操作符的Flow：buffer、conflate、flowIn、shareIn。

收集单个Flow使用	Flow.flowWithLifecyle( )
收集多个流、热流使用	Lifecyle.repeatOnLifecyle( )
不搜集流使用	LifecycleScope.launchWhenX( )

3.1 Flow.flowWithLifecyle( )

当只有一个Flow需要收集时使用该操作符，内部基于Lifecyle.repeatOnLifecyle()实现的扩展函数。生命周期低于目标状态会取消上游，不影响下游，需要注意调用顺序。

public fun <T> Flow<T>.flowWithLifecycle(
lifecycle: Lifecycle,
minActiveState: Lifecycle.State = Lifecycle.State.STARTED
): Flow<T> = callbackFlow {
lifecycle.repeatOnLifecycle(minActiveState) {
this@flowWithLifecycle.collect {
send(it)
}
}
close()
}

3.2 Lifecyle.repeatOnLifecyle( )

在生命周期到达目标状态时，挂起调用它的协程。
会在离开目标状态时取消子协程，再次进入目标状态会重开子协程，即围绕目标状态的进出多次重新执行代码。
进入Destroy状态才会恢复调用它的协程。

public suspend fun Lifecycle.repeatOnLifecycle(
state: Lifecycle.State,//可选状态：INITIALIZED、CREATED、STARTED、RESUMED、DESTROYED
block: suspend CoroutineScope.() -> Unit
)

lifecycleScope.launch { //会挂起之前的代码 repeatOnLifecycle(Lifecycle.State.STARTED) { //重复的工作 } //当进入onDestroy()状态协程会恢复执行后面的代码 }

3.3LifecycleScope.launchWhenX( )

会在进入X状态前等待、进入后执行、离开后挂起。如果内部订阅了数据流，挂起只是停止消费数据，没有取消协程，无法阻止被订阅的数据流继续活跃生产数据，由于会造成资源浪费已被废弃，使用上面的 3.2 替代。

public fun launchWhenCreated(block: suspend CoroutineScope.() -> Unit): Job
public fun launchWhenResumed(block: suspend CoroutineScope.() -> Unit): Job
public fun launchWhenStarted(block: suspend CoroutineScope.() -> Unit): Job