如何实现Kotlin Flow定期轮询刷新并持续监听Room数据库变化

问题描述 #

• 在Room数据库中存储了书籍ID列表，可通过向网络API传入已存储的ID，获取对应的实际书籍数据
• 初始目标为监听数据库变化，将书籍ID映射为实际书籍对象，这部分逻辑目前已可正常运行
• 需求：在全程保持数据库监听的前提下，定期从网络API重新拉取书籍数据；数据库中的书籍ID发生变更时，必须始终触发网络API调用booksApi.get()
  当前仅提供思路的非可运行实现代码如下：

fun getBookesFlow(): Flow<List<Book>> {
  return booksDao.all().map { ids -> 
    booksApi.get(ids)
  }
}

fun getPollingFlow(): Flow<List<Book>> {
  return flow {
    while(true) {
      // 每30秒重置一次Flow
      getBookesFlow().collect { bookes -> // 仅为思路示意，非可运行代码
        emit(books)
      }
      delay(30.seconds)
    }
  }
}

fun getBooks() {
  viewModelScope.launch {
   getPollingFlow().collect { books ->
    _uiState.update { it.copy( books = books) }
   }
  }
}

解决方案 #

原代码无法运行的核心原因：第一次进入循环后，协程会一直挂起在getBookesFlow().collect的持续收集操作上，永远走不到后续的delay(30.seconds)逻辑，定时刷新逻辑自然不会生效。

直接使用combine操作符合并「数据库ID变更流」和「定时触发流」即可，两个流中任意一个发出新值，都会触发一次网络请求，完全匹配需求：

import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.flow
import kotlin.time.Duration.Companion.seconds

fun getBooksFlow(): Flow<List<Book>> {
    // 定义30秒间隔的定时触发流，启动时立即触发一次拉取
    val refreshTimer = flow {
        while (true) {
            emit(Unit)
            delay(30.seconds)
        }
    }

    return combine(
        booksDao.all(),
        refreshTimer
    ) { latestIds, _ ->
        // 不管是ID变更还是定时触发，都用最新的ID列表拉取网络数据
        booksApi.get(latestIds)
    }
}

// 视图层调用逻辑无需改动
fun getBooks() {
  viewModelScope.launch {
   getBooksFlow().collect { books ->
    _uiState.update { it.copy(books = books) }
   }
  }
}

方案特性 #

• 数据库ID发生变更时，Room的观察流会立即推送最新ID列表，combine第一时间触发网络请求，不会漏掉任何数据库变更
• 每间隔30秒，定时流发出触发信号，combine会读取当前最新的数据库ID列表，主动拉取一次最新的网络数据
• 流的生命周期完全和下游收集协程绑定，ViewModel销毁时协程自动取消，不会产生内存泄漏

如果需要避免屏幕旋转等配置变更导致流重启、重复发起网络请求，可以通过stateIn将流转换为带缓存的共享StateFlow：

private val booksState = getBooksFlow()
    .stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5000),
        initialValue = emptyList()
    )

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