如何在取消前收集防抖Kotlin Flow的最新值？

我们在使用Kotlin Flow的debounce操作符处理高频发射流时，会遇到协程作用域被取消后丢失最新未触发debounce值的问题。比如提前取消作用域时，本该收集到的中间值会被丢弃；同时要避免在debounce已经完成发射后，取消时重复发送已收集的值。

解决方案：自定义扩展函数 #

我们可以通过组合debounce、状态保存和取消监听来实现需求，以下是封装好的扩展函数：

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.debounce
import kotlinx.coroutines.flow.onCompletion
import java.util.concurrent.atomic.AtomicBoolean

fun <T> Flow<T>.debounceWithFinalEmit(timeoutMillis: Long): Flow<T> {
    // 保存最新发射的值
    val latestValue = MutableStateFlow<T?>(null)
    // 标记最新值是否已经通过debounce发射过
    val isLatestEmitted = AtomicBoolean(false)

    return this
        // 每次收到新值时更新最新值，并重置发射标记
        .onEach {
            latestValue.value = it
            isLatestEmitted.set(false)
        }
        // 原有debounce逻辑
        .debounce(timeoutMillis)
        // debounce发射值时，标记该值已被处理
        .onEach {
            isLatestEmitted.set(true)
        }
        // 监听流完成事件，若因取消导致完成，且最新值未被发射，则发送该值
        .onCompletion { cause ->
            if (cause is CancellationException) {
                latestValue.value?.takeIf { !isLatestEmitted.get() }?.let { emit(it) }
            }
        }
}

测试场景验证 #

场景1：300ms后取消作用域

val scope = CoroutineScope(Dispatchers.IO)
scope.launch {
    flow { for (i in 1 until 5) { emit(i); delay(200.milliseconds) } }
        .onEach { println("emitted $it") }
        .debounceWithFinalEmit(300.milliseconds)
        .collect { println("collected $it") }
}

Thread.sleep(300)
scope.cancel()
Thread.sleep(1000)

输出结果：

emitted 1
emitted 2
collected 2

场景2：1100ms后取消作用域（值4已通过debounce发射）

val scope = CoroutineScope(Dispatchers.IO)
scope.launch {
    flow { for (i in 1 until 5) { emit(i); delay(200.milliseconds) } }
        .onEach { println("emitted $it") }
        .debounceWithFinalEmit(300.milliseconds)
        .collect { println("collected $it") }
}

Thread.sleep(1100)
scope.cancel()
Thread.sleep(1000)

输出结果：

emitted 1
emitted 2
emitted 3
emitted 4
collected 4

实现原理 #

1. 状态保存：用MutableStateFlow实时记录流发射的最新值，确保取消时能获取到未处理的最后一个值。
2. 发射标记：通过AtomicBoolean标记最新值是否已经被debounce正常发射，避免取消时重复发送已处理的值。
3. 取消监听：在onCompletion中判断流是否因取消而结束，若满足条件且最新值未被发射，则立即发送该值，解决丢失问题。

内容的提问来源于stack exchange，提问作者Robin