如何获取Android中Kotlin SharedFlow发射的最新值？

我在使用Kotlin SharedFlow 时，尝试用 last() 运算符获取流发射的最新值，但没达到预期效果。

初始代码 #

main函数代码 #

fun main(): Unit = runBlocking {
    val eventBus = EventBus()
    launch {
        val latestValue = eventBus.value.lastOrNull()
        if (latestValue != null) {
            println("Latest value: $latestValue")
        }
    }
    launch {
        repeat(5) {
            eventBus.increment(it)
        }
    }
    launch {
        delay(5.seconds)
        (5..10).forEach {
            eventBus.increment(it)
        }
    }
}

EventBus类代码 #

class EventBus {
    private val _value = MutableSharedFlow<Int>()
    val value: SharedFlow<Int> = _value.asSharedFlow()

    suspend fun increment(number: Int) {
        _value.emit(number)
    }
}

需求说明 #

我希望捕获流发射的最新值：比如流先发射0到4，暂停一段时间后再发射5到10，我想分别保留每个区间的最终值——初始区间保留4，第二个区间保留10，这能实现吗？

更新后的尝试 #

按照建议修改代码后，还是无法打印第二个区间的最新值：

fun main(): Unit = runBlocking {
    val eventBus = EventBus()
    launch {
        repeat(5) {
            eventBus.increment(it)
        }
    }
    launch {
        delay(5.seconds)
        (5..10).forEach {
            delay(100)
            eventBus.increment(it)
        }
    }
    launch {
        val latestValue = eventBus.value.replayCache.firstOrNull()
        if (latestValue != null) {
            println("Latest value: $latestValue")
        }
    }
}

class EventBus {
    private val _value = MutableSharedFlow<Int>(replay = 1)
    val value: SharedFlow<Int> = _value.asSharedFlow()

    suspend fun increment(number: Int) {
        _value.emit(number)
    }
}

问题原因与解决方案 #

1. 核心问题分析 #

• 初始代码中，默认的MutableSharedFlow没有重播缓存（replay = 0），调用lastOrNull()会一直挂起等待流结束，但你的SharedFlow是永不结束的，因此协程会一直阻塞，错过所有发射的值。
• 修改replay=1后，仅读取一次replayCache.firstOrNull()只能拿到当前时刻的缓存值，无法监听后续的流更新，自然获取不到第二个区间的10。

2. 实现需求的三种方案 #

方案一：持续监听并记录最新值

通过collect持续监听流，实时更新并记录最新值，还可针对性捕获区间结束值：

fun main(): Unit = runBlocking {
    val eventBus = EventBus()
    val latestValue = mutableStateOf<Int?>(null)
    
    // 持续监听流，更新最新值并按需打印区间结束值
    launch {
        eventBus.value.collect {
            latestValue.value = it
            if (it == 4 || it == 10) {
                println("区间最终值: $it")
            }
        }
    }

    launch {
        repeat(5) {
            eventBus.increment(it)
        }
    }

    launch {
        delay(5.seconds)
        (5..10).forEach {
            delay(100)
            eventBus.increment(it)
        }
    }

    // 主动获取指定时间点的最新值
    delay(6.seconds)
    println("当前最新值: ${latestValue.value}")
}

class EventBus {
    private val _value = MutableSharedFlow<Int>(replay = 1)
    val value: SharedFlow<Int> = _value.asSharedFlow()

    suspend fun increment(number: Int) {
        _value.emit(number)
    }
}

方案二：用运算符过滤区间结束值

如果明确知道区间的结束值，可通过filter+first精准捕获：

fun main(): Unit = runBlocking {
    val eventBus = EventBus()
    
    launch {
        // 等待第一个区间的结束值4
        val firstLast = eventBus.value.filter { it == 4 }.first()
        println("第一个区间最终值: $firstLast")
        // 等待第二个区间的结束值10
        val secondLast = eventBus.value.filter { it == 10 }.first()
        println("第二个区间最终值: $secondLast")
    }

    launch {
        repeat(5) {
            eventBus.increment(it)
        }
    }

    launch {
        delay(5.seconds)
        (5..10).forEach {
            delay(100)
            eventBus.increment(it)
        }
    }
}

class EventBus {
    private val _value = MutableSharedFlow<Int>(replay = 1)
    val value: SharedFlow<Int> = _value.asSharedFlow()

    suspend fun increment(number: Int) {
        _value.emit(number)
    }
}

方案三：改用StateFlow（最适合获取最新值场景）

如果核心需求是获取和监听最新值，StateFlow是更简洁的选择——它天生持有当前最新状态值，无需额外处理缓存：

class EventBus {
    private val _value = MutableStateFlow<Int?>(null) // 可设初始值为null或默认值
    val value: StateFlow<Int?> = _value.asStateFlow()

    fun increment(number: Int) {
        _value.value = number // StateFlow赋值无需挂起，也可用emit
    }
}

fun main(): Unit = runBlocking {
    val eventBus = EventBus()
    
    launch {
        // 监听最新值变化，按需打印区间结束值
        eventBus.value.collect {
            if (it != null && (it == 4 || it == 10)) {
                println("区间最终值: $it")
            }
        }
    }

    launch {
        repeat(5) {
            eventBus.increment(it)
        }
    }

    launch {
        delay(5.seconds)
        (5..10).forEach {
            delay(100)
            eventBus.increment(it)
        }
    }

    // 主动获取指定时间点的最新值
    delay(6.seconds)
    println("当前最新值: ${eventBus.value.value}")
}

关键总结 #

• SharedFlow的last()会等待流完成，若流永不结束则不会返回结果。
• 设置replay=1后，需持续监听流才能获取后续更新的最新值，单次读取缓存无法满足需求。
• 若核心需求是获取和监听最新值，StateFlow是更适配的选择，API更简洁直观。

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