如何在无内存泄漏、无生命周期场景下从闭包中收集流数据？

我们可以手动实现一个具备初始值、支持订阅且自动管理内存的流类型，完全不依赖第三方库，核心通过弱引用管理订阅者+手动可取消的订阅凭证来避免内存泄漏，适配无协程、无生命周期的场景需求。

1. 定义核心接口 #

先明确对外暴露的能力：获取当前值、订阅值变化，定义两个基础接口：

interface StreamWithInitialValue<T> {
    val currentValue: T
    fun sink(onValue: (T) -> Unit): Disposable
}

interface Disposable {
    fun dispose()
}

2. 实现具体流类 #

实现DefaultStreamWithInitialValue，内部用弱引用保存订阅者，确保订阅者被回收时自动清理无效订阅；同时提供更新值并通知订阅者的能力：

class DefaultStreamWithInitialValue<T>(initialValue: T) : StreamWithInitialValue<T> {
    override var currentValue: T = initialValue
        private set

    // 用弱引用集合存储订阅者，避免内存泄漏
    private val subscribers = mutableSetOf<WeakReference<Subscriber<T>>>()

    override fun sink(onValue: (T) -> Unit): Disposable {
        val subscriber = Subscriber(onValue)
        val weakRef = WeakReference(subscriber)
        subscribers.add(weakRef)
        // 订阅时立即触发一次当前值，和CurrentValueSubject行为对齐
        onValue(currentValue)
        // 返回订阅凭证，支持手动取消
        return object : Disposable {
            override fun dispose() {
                subscribers.remove(weakRef)
            }
        }
    }

    // 更新值并通知所有活跃订阅者
    fun setValue(newValue: T) {
        if (currentValue == newValue) return
        currentValue = newValue
        // 先清理已被回收的弱引用，再通知有效订阅者
        subscribers.removeIf { it.get() == null }
        subscribers.forEach { it.get()?.onValue(newValue) }
    }

    // 内部订阅者包装类
    private class Subscriber<T>(private val onValue: (T) -> Unit) {
        fun onValue(value: T) = onValue.invoke(value)
    }
}

3. 使用示例 #

完全贴合你给出的场景，无需协程和生命周期，通过持有Disposable管理订阅生命周期，当持有对象被回收时，订阅会自动失效：

class ExampleClass {
    private var someStrongReference: Disposable? = null

    fun listen(stream: StreamWithInitialValue<Boolean>) {
        val now = stream.currentValue
        println("当前初始值: $now")

        someStrongReference = stream.sink { newBoolean ->
            println("收到新值: $newBoolean")
        }
    }

    // 可选：主动取消订阅
    fun stopListening() {
        someStrongReference?.dispose()
        someStrongReference = null
    }
}

// 测试代码
fun main() {
    val stream = DefaultStreamWithInitialValue(false)
    val example = ExampleClass()
    example.listen(stream)

    stream.setValue(true)
    stream.setValue(false)

    // 当example被GC回收时，someStrongReference也会被回收，订阅自动失效
}

关键特性说明 #

• 无协程依赖：全程采用回调式实现，不依赖任何协程API
• 自动内存管理：通过WeakReference存储订阅者，订阅者被回收时自动清理无效订阅，避免内存泄漏
• 初始值触发：订阅时立即回调当前值，和Combine的CurrentValueSubject行为一致
• 灵活生命周期控制：支持手动取消订阅，也可通过持有对象的回收自动终止订阅

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