Room 与 Flow 集成：用 Flow 做 DAO 查询返回

记得我刚接触 Room + Flow 那会儿，第一反应是“这不就是把 LiveData 换成 Flow 吗？”
后来踩了几个坑才明白——差别远不止名字不同。

Room 对 Flow 的支持，本质上是把数据库的响应式能力提升了一个档次。你写一个 @Query 方法，返回类型声明成 Flow<List<User>>，Room 就会自动帮你做到：只要有表数据发生变化，新结果就会通过 Flow 发射出来。不用手动调用、不用观察 Lifecycle、不用考虑取消——因为 Flow 的冷流特性加上协程作用域，天然适合后台数据订阅。

基础姿势：从 LiveData 迁移到 Flow

先看个简单的例子。假设有个 UserDao：

@Dao
interface UserDao {
    @Query("SELECT * FROM users ORDER BY name ASC")
    fun getAllUsersFlow(): Flow<List<User>>

    @Insert
    suspend fun insert(user: User)

    @Delete
    suspend fun delete(user: User)
}

注意到没？函数名加了个 Flow 后缀——这是我个人的命名习惯，能一眼看出它是响应式的。调用方只需要这样收集：

viewModelScope.launch {
    userDao.getAllUsersFlow().collect { users ->
        // 每次 users 表变化，这里都会收到最新数据
    }
}

我项目中遇到过几次这样的场景：数据库里有一张配置表，需要实时响应远端推送的变更。以前用 LiveData 要手动加 observe、还要在 Fragment 里处理生命周期。换成 Flow 后，直接在 Repository 层用 flow.catch 处理异常，干净利落。

💡 小提示：返回 Flow 的 @Query 方法，Room 内部是通过 InvalidationTracker 实现的。它监听所有涉及的表，任何行增删改都会触发重新查询。所以不用写多余的触发器。

Flow 查询的内部机制：不仅仅是“自动刷新”

为什么 Room 能做到自动推送？其实就是一张图：

Room + Flow 响应式查询流程 @Query 返回 Flow SQLite 执行查询 结果集 → Flow 发射 InvalidationTracker 监听表变化 自动重查 → 再次发射 整个过程由 Room 自动管理，开发者只需关注 collect 中的数据

流程其实很清晰：

  1. 你调用 DAO 方法，Room 先执行一次 SQLite 查询。
  2. 结果包装成 Flow 发射出去。同时 InvalidationTracker 开始监听涉及的表。
  3. 任何 INSERT、UPDATE、DELETE 后，Room 检测到表变动，自动重新查询并发射新结果。

嗯，这里要注意：重查是全量查询，不是增量。如果你查询了 10 万条数据，那么每次表变动都会重新查 10 万条。所以查询本身必须快——索引要建好、SQL 要优化。我曾经在一个历史数据表上直接返回 Flow，没加 LIMIT，结果每次插入一条记录就要查几十万行，UI 直接卡死。

避坑指南：那些你容易踩的坑

⚠️ 常见陷阱
  • Flow + 大表 = 性能炸弹：全量重查，数据量大时请加分页或LIMIT。
  • 多个表关联查询时：Room 的 InvalidationTracker 只能监听单个 SQL 中涉及的表。如果用了JOIN，只要任一关联表变化就会重查，可能触发不必要的重查。
  • 跨数据库变更不感知：Room 只监控自己的 SQLite 数据库。如果你通过其他方式（比如直接写 db 文件）改了数据，Flow 不会被触发。

什么时候用 Flow？什么时候该换回 suspend？

我个人的经验是：

场景推荐方式原因
列表页面需要实时刷新Flow自动推送，不用手动轮询
单个详情页（只加载一次）suspend fun避免不必要的订阅，减少开销
跨页面共享数据快照StateFlow + Room Flow配合 stateIn 转为热流
聚合查询、报表Flow + 去抖高频写入时加 debounce 减少重查次数
📌 核心原则：需要实时响应变化就用 Flow，只需要一次快照就用 suspend。
我见过不少团队把整个 DAO 都改成返回 Flow，结果导致某些一次性查询多次触发——说白了，不是所有的数据都需要“活着”。

高级技巧：用 Flow 做数据转换

Room 返回的 Flow 是冷流，每次重查都是新的一次 SQL 查询。你可以在收集前用 Flow 操作符做转换：

// DAO
@Query("SELECT * FROM users WHERE is_active = 1")
fun getActiveUsers(): Flow<List<User>>

// Repository
fun getActiveUsersNameList(): Flow<List<String>> {
    return userDao.getActiveUsers()
        .map { users -> users.map { it.name } }
        .catch { e -> Log.e("DB", "查询异常", e); emit(emptyList()) }
}

你会问：为什么不直接在 SQL 里只查 name？我实际项目中遇到过这样的场景：既需要 name，又需要判断用户角色以做后续逻辑。如果只查 name，第二次需求变更时又要改 SQL。所以我的习惯是——DAO 层尽量返回完整实体，上层用 Flow 操作符做裁剪。这样 SQL 复用度高，调试也更方便。

另外，如果你在 ViewModel 里做 stateIn 转换：

val userList: StateFlow<List<User>> = userDao.getAllUsersFlow()
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

注意 WhileSubscribed(5000) 这个参数。我刚开始直接用 Eagerly，结果一启动就订阅数据库，即使 Activity 还没初始化。后来换成 WhileSubscribed 加上合理超时（5秒），既避免了频繁重建 streaming，又不会在后台浪费资源。

最后聊聊「一致性验证」

Flow 本身不保证数据一致性——它只保证"当你收集到时，数据是某个时刻的快照"。在回归测试中，你核心要验证的是：

  • 表变化后 Flow 是否重新发射（通过 CountDownLatch 或 kotlinx.coroutines.test 等待多个值）
  • 发射的数据是否与最新数据库状态一致（对比 Room 的同步查询结果）
  • 并发写入时 Flow 的发射间隔是否合理（避免丢数据或重复发射）

这部分会在后面的章节详细展开，但今天你记住一点：Flow 是数据的“搬运工”，不是“质检员”。验证工作需要你自己在测试代码里完成。

最后留个小问题给你想想：如果数据库同时被多个线程写入，Room 的 InvalidationTracker 是怎么保证不丢失事件的？它用的是 CountDownLatch 还是锁？答案我在后面的章节会讲。

← 上一章 📖 返回目录 下一章 →