RxAndroidBle大量写操作生成过多Disposable的内存与实现问题咨询

内存占用问题的原因 #

当前代码每次调用writeBle都会创建新的订阅并加入connectionDisposable，即使写操作完成或失败，对应的Disposable仍会留在CompositeDisposable中。虽然这些Disposable已经处于disposed状态，但大量累积仍会占用内存，尤其是当调用次数达到数千次时，会导致不必要的内存开销。

内存问题的快速修复 #

如果要保留现有调用方式，可在订阅完成或出错时主动从CompositeDisposable中移除对应的Disposable：

private fun writeBle(writeCharacteristicUuid: UUID, command: ByteArray) {
    if (bleDevice.connectionState == RxBleConnection.RxBleConnectionState.CONNECTED) {
        val disposable = activeConnection
            .flatMapSingle {
                it.writeCharacteristic(writeCharacteristicUuid, command)
            }
            .subscribe(
                {
                    Log.d(TAG, "${connectionDisposable.size()} - Command successful: ${it.toHexString()}")
                    connectionDisposable.remove(disposable) // 完成后移除
                },
                {
                    Log.e(TAG, "Error executing command: $it")
                    connectionDisposable.remove(disposable) // 出错后移除
                }
            )
        connectionDisposable.add(disposable)
    } else {
        Log.e(TAG, "You are not connected")
    }
}

不过这种方式只是临时修复，无法从根本上解决大量命令累积的问题，更优的方案是采用队列化的命令处理方式。

大量非同时写命令的正确实现 #

BLE协议要求写操作必须串行执行（不能同时发送多个写命令），因此最佳实践是用一个队列来管理所有写请求，确保命令按顺序执行，同时避免大量Disposable累积：

1. 定义命令队列Subject #

private val writeCommandSubject = PublishSubject.create<Pair<UUID, ByteArray>>()
private var writeQueueDisposable: Disposable? = null

2. 初始化队列订阅（在连接建立后调用） #

private fun setupWriteQueue(connection: RxBleConnection) {
    writeQueueDisposable?.dispose() // 先清理旧的订阅
    writeQueueDisposable = writeCommandSubject
        .concatMapSingle { (uuid, command) ->
            connection.writeCharacteristic(uuid, command)
        }
        .subscribe(
            { Log.d(TAG, "Command successful: ${it.toHexString()}") },
            { Log.e(TAG, "Command failed: $it") }
        )
}

这里使用concatMapSingle确保每个写命令在前一个完成后才会执行，严格串行化。

3. 修改writeBle方法，将命令发送到队列 #

private fun writeBle(writeCharacteristicUuid: UUID, command: ByteArray) {
    if (bleDevice.connectionState == RxBleConnection.RxBleConnectionState.CONNECTED) {
        writeCommandSubject.onNext(writeCharacteristicUuid to command)
    } else {
        Log.e(TAG, "You are not connected")
    }
}

4. 连接关闭时清理资源 #

private fun onConnectionClosed() {
    writeQueueDisposable?.dispose()
    writeCommandSubject.onComplete() // 可选，标记队列完成
    connectionDisposable.clear()
}

方案优势 #

• 整个写流程仅维护一个Disposable，彻底解决大量Disposable累积的内存问题
• 自动保证命令串行执行，符合BLE协议要求，避免并发写导致的错误
• 代码结构更清晰，命令发送与执行逻辑分离

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