基于Kotlin Flows实现HashMap更新订阅的优化方案问询

数据老陈

2026-7-5

问题背景与实现探讨

先看最初的数据源实现：

data class Message(val id: String, val text: String)
data class Conversation(val id: String, val messages: List<Message>)
class InMemoryDataSource {
    private val conversations: MutableMap<String, Conversation> = mutableMapOf()

    fun updateConversation(conversationID: String, newMessage: Message) {
        conversations[conversationID]?.let { currentConversation ->
            conversations[conversationID] = currentConversation.copy(
                messages = currentConversation.messages + newMessage
            )
        }
    }
}

为了支持订阅整个会话集合和单个会话的更新，我实现了带Flow的版本：

class ObservableInMemoryDataSource {
    val conversationsFlow = MutableStateFlow<MutableMap<String, MutableStateFlow<Conversation>>>(mutableMapOf())

    fun updateConversation(conversationID: String, newMessage: Message) {
        conversationsFlow.update { conversationsMap ->
            conversationsMap[conversationID]?.update { conversation ->
                conversation.copy(
                    messages = conversation.messages + newMessage
                )
            } ?: run {
                conversationsMap[conversationID] =
                    MutableStateFlow(Conversation(id = conversationID, listOf(newMessage)))
            }
            conversationsMap
        }
    }
}

对应的示例运行代码：

fun main(args: Array<String>) {
    runBlocking {
        val source = ObservableInMemoryDataSource()
        launch {
            source.conversationsFlow.collect {
                println("--------updates from the conversationsFlow--------")
                it.forEach {conversationFlow ->
                    println(conversationFlow.value.value)
                }

            }
        }
        source.updateConversation("1", Message("1", "hello"))
        delay(1000)
        launch {
            source.conversationsFlow.value["1"]?.collect{
                println("--------updates from the conversation[1]--------")
                println(it)
            }
        }
        source.updateConversation("1", Message("2", "one more"))
        println("should be updating")
    }
}

这个实现能正常工作，但我有三个核心担忧：

嵌套大量Flow的性能问题
删除会话时，如何避免已移除的单个Flow实例引发内存泄漏
并发安全性：是否需要用Mutex保证线程安全

优化后的实现（编辑#1）

我自己调整了一个更简洁的版本：

class InMemoryDataSource {
    private val _conversations = MutableStateFlow<Map<String, Conversation>>(emptyMap())
    val conversations: Flow<Map<String, Conversation>> = _conversations.asStateFlow()
    private val mutex = Mutex()

    fun getConversation(conversationId: String): Flow<Conversation?> {
        return _conversations.asStateFlow().map { it[conversationId] }
    }

    suspend fun updateConversation(conversationID: String, newMessage: Message) = mutex.withLock {
        _conversations.update { state ->
            val result = state[conversationID]?.let { currentConversation ->
                state
                    .filterNot { it.key == conversationID }
                    .plus(
                        mapOf(
                            conversationID to currentConversation.copy(
                                messages = currentConversation.messages + newMessage
                            )
                        )
                    )
            } ?: run {
                state + mapOf(conversationID to Conversation(id = conversationID, listOf(newMessage)))
            }
            result
        }
    }

    fun removeConversation(conversationID: String) = mutex.withLock {
        _conversations.update {
            it.filterNot { it.key == conversationID }
        }
    }
}

针对三个担忧的分析与建议

1. 嵌套Flow的性能问题

最初的嵌套MutableStateFlow实现存在几个性能隐患：

单个会话更新时，外层conversationsFlow会因持有可变Map的修改触发更新，导致所有订阅集合的观察者都收到通知，即便只有一个会话变化。
大量独立StateFlow会增加内存开销和订阅管理复杂度，会话数量大时内存占用会显著上升。

你优化后的方案更高效：

单个StateFlow分发状态，自带去重逻辑，仅在状态真正变化时通知观察者。
单个会话的Flow由主Flow派生而来，无需维护大量独立Flow，内存开销更低，订阅逻辑更统一。

2. 删除会话时的内存泄漏问题

嵌套Flow方案中，若移除会话的Flow但仍有观察者订阅，该Flow实例无法被GC回收，会引发内存泄漏。

优化后的版本完美解决了这个问题：

单个会话的Flow是主状态的映射，会话删除后派生Flow会自动发射null，无额外无效引用持有；即便观察者未取消订阅，也不会留存孤立的Flow实例。
所有状态集中在主StateFlow，删除操作仅修改主状态，无残留资源。

3. 并发安全性

必须使用Mutex保证线程安全：

MutableStateFlow.update本身线程安全，但多步骤的更新逻辑（读状态→修改→更新）在并发场景下会出现竞态条件。
你优化后的版本用mutex.withLock包裹修改逻辑，确保同一时间只有一个线程修改主状态，避免状态不一致。

注意：原removeConversation方法需改为挂起函数，因为mutex.withLock是挂起函数，否则会编译报错，修正后：

suspend fun removeConversation(conversationID: String) = mutex.withLock {
    _conversations.update {
        it.filterNot { it.key == conversationID }
    }
}

额外健壮性建议

坚持使用不可变数据结构：主Flow用不可变Map，所有状态更新生成新Map，保证状态可追溯与线程安全。
避免无效状态分发：若更新逻辑可能生成重复状态（如重复添加相同Message），可在update前做判断，减少不必要的通知。
提醒订阅管理：告知使用者在不需要时取消订阅（如用lifecycleScope或手动调用Job.cancel()），避免资源浪费。
增加错误处理：可在更新方法中返回Result<Unit>或添加异常捕获，让调用方知晓更新结果。

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