如何将基于注册的外部API转换为Kotlin SharedFlow并移除监听器？

要解决这个问题，核心是在Flow不再被订阅或被取消时，自动移除注册到外部API的监听器，避免内存泄漏。推荐使用callbackFlow来实现桥接，它专门适配回调/监听器风格的API，并且内置了流关闭时的清理机制。

最终实现代码 #

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.stateIn

fun API.toFlow(): SharedFlow<Int> = callbackFlow {
    // 创建监听器，收到外部API更新时发送到Flow
    val listener = Listener { newValue ->
        trySend(newValue)
    }
    // 注册监听器到外部API
    this@toFlow.listen(listener)

    // 核心逻辑：当流关闭时（无订阅者/被取消）自动移除监听器
    awaitClose {
        this@toFlow.stopListening(listener)
    }
}
// 转换为SharedFlow，支持多订阅者并配置活跃策略
.stateIn(
    scope = CoroutineScope(Dispatchers.Default),
    started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5000),
    initialValue = 0 // 根据业务需求调整初始值，不需要可替换为合适默认值
)

关键细节说明 #

• callbackFlow的作用：它是Kotlin Flow专为回调转Flow场景设计的API，awaitClose块中的代码会在流的最后一个订阅者取消订阅、流被取消或完成时执行，完美匹配我们需要移除监听器的时机。
• stateIn的配置：
  • SharingStarted.WhileSubscribed(stopTimeoutMillis = 5000)：当没有订阅者时，等待5秒后停止流（触发awaitClose清理监听器）；如果在超时前有新订阅者，流会立即恢复活跃，平衡资源消耗和响应速度。
  • 如果需要流始终保持活跃（比如后台持续监听），可以把started参数改成SharingStarted.Eagerly，但会持续占用资源，按需选择。
• 避免内存泄漏：通过awaitClose确保监听器在流不再被使用时被移除，不会一直绑定到外部API上。

原方案的问题 #

你原来的代码直接创建MutableSharedFlow并注册监听器，但没有任何机制感知Flow的订阅状态或生命周期，监听器会一直留在外部API中，导致内存泄漏，并且无法在合适时机停止监听。

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