探究Flow与Room、SQLite底层实现，原生SQLite的Flow方案改造

Room中Flow响应数据变更的底层原理 #

Room确实通过内置的数据库观测机制实现了Flow的数据变更响应：

• Room在编译期会为返回Flow的查询方法生成代理代码，自动为目标表注册变更监听。
• 当数据库执行插入、更新、删除等写操作时，Room会检测到目标表的变化，触发查询重新执行，并将最新的查询结果通过Flow发射给订阅者。
• 底层依赖SQLite的WAL（预写日志）机制追踪变更，结合自定义的监听逻辑实现数据更新的自动通知，无需开发者手动处理监听与通知逻辑。

基于原生SQLite实现Flow的方案 #

原生SQLite没有内置的变更通知机制，我们需要手动实现监听逻辑，结合协程的callbackFlow将查询结果转换为可观测的Flow。以下是两种可行的实现方式：

方式1：借助ContentProvider与ContentObserver #

如果你的数据库通过ContentProvider暴露，可利用系统的ContentObserver监听变更：

import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.callbackFlow
import android.content.ContentObserver
import android.os.Handler
import android.os.Looper
import android.net.Uri

// 假设你的ContentProvider对应的Uri
private val USER_CONTENT_URI = "content://com.your.app.provider/user".toUri()

fun getAllFlow(context: Context): Flow<List<User>> = callbackFlow {
    // 创建数据库变更监听器
    val changeObserver = object : ContentObserver(Handler(Looper.getMainLooper())) {
        override fun onChange(selfChange: Boolean) {
            // 变更触发时重新查询并发送数据
            val latestUsers = getAll() ?: emptyList()
            trySend(latestUsers)
        }
    }

    // 注册监听器
    context.contentResolver.registerContentObserver(USER_CONTENT_URI, true, changeObserver)

    // 发送初始数据
    val initialUsers = getAll() ?: emptyList()
    trySend(initialUsers)

    // Flow取消时注销监听器，避免内存泄漏
    awaitClose {
        context.contentResolver.unregisterContentObserver(changeObserver)
    }
}

// 原查询方法调整为返回非空列表
fun getAll(): List<User> {
    val sql = "SELECT id, first, last FROM user"
    val users = mutableListOf<User>()
    connection.readableDatabase.rawQuery(sql, null).use { cursor ->
        while (cursor.moveToNext()) {
            users.add(User(cursor))
        }
    }
    return users
}

方式2：手动维护变更监听器列表 #

如果不使用ContentProvider，可在SQLiteOpenHelper子类中维护监听器，写操作后主动通知：

import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.callbackFlow
import android.content.Context
import android.database.sqlite.SQLiteOpenHelper
import android.content.ContentValues

class MyDatabaseHelper(context: Context) : SQLiteOpenHelper(context, "user_db", null, 1) {
    private val changeListeners = mutableList<() -> Unit>()

    // 注册变更监听器
    fun addChangeListener(listener: () -> Unit) = synchronized(changeListeners) {
        changeListeners.add(listener)
    }

    // 注销变更监听器
    fun removeChangeListener(listener: () -> Unit) = synchronized(changeListeners) {
        changeListeners.remove(listener)
    }

    // 通知所有监听器数据库已变更
    private fun notifyDatabaseChanged() = synchronized(changeListeners) {
        changeListeners.forEach { it.invoke() }
    }

    // 示例插入方法，执行后触发通知
    fun insertUser(user: User) {
        writableDatabase.use { db ->
            db.insert("user", null, ContentValues().apply {
                put("first", user.first)
                put("last", user.last)
            })
        }
        notifyDatabaseChanged()
    }

    // 更新、删除方法同理，执行后调用notifyDatabaseChanged
}

// 实现Flow
fun getAllFlow(dbHelper: MyDatabaseHelper): Flow<List<User>> = callbackFlow {
    val changeListener = {
        val latestUsers = getAll(dbHelper)
        trySend(latestUsers)
    }

    dbHelper.addChangeListener(changeListener)

    // 发送初始数据
    val initialUsers = getAll(dbHelper)
    trySend(initialUsers)

    // Flow取消时注销监听器
    awaitClose {
        dbHelper.removeChangeListener(changeListener)
    }
}

fun getAll(dbHelper: MyDatabaseHelper): List<User> {
    val sql = "SELECT id, first, last FROM user"
    val users = mutableListOf<User>()
    dbHelper.readableDatabase.rawQuery(sql, null).use { cursor ->
        while (cursor.moveToNext()) {
            users.add(User(cursor))
        }
    }
    return users
}

核心要点说明 #

• callbackFlow是将回调逻辑转换为Flow的核心API，它能安全处理协程的生命周期，确保资源正确释放。
• 无论哪种方式，都需要保证写操作后触发变更通知，否则Flow无法感知数据变化。
• 监听器的注册与注销必须成对出现，awaitClose会在Flow取消时自动执行注销逻辑，避免内存泄漏。

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