ViewModel 中 StateFlow 与 SharedFlow 的使用建议及单元测试指南

参数说明：

• scope: 通常传入 viewModelScope，确保生命周期绑定。
• initialValue: 初始状态，UI 加载时立即显示。
• started: 控制何时开始收集。WhileSubscribed 表示当没有订阅者时停止收集，并在超时后清理，防止内存泄漏。

多数据源合并

在一些业务复杂的页面，比如首页，通常会有多个数据来源。为了保证单一可靠数据源原则，我们可以使用 combine 函数将多个 flow 组成一个 flow，然后再使用 stateIn 函数转换成 StateFlow。

val combinedState = combine(userFlow, settingsFlow) { user, settings ->
    CombinedState(user, settings)
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(), CombinedState.EMPTY)

StateFlow 与 SharedFlow 如何选择？

虽然两者都是热流，但在具体场景下选择有所不同。选择时主要考虑以下问题：

1. 我真的需要在特定的时间、位置获取 Flow 的最新状态吗？

   • 如果不需要（例如事件通知），那考虑 SharedFlow。
   • 如果需要（例如 UI 状态），StateFlow 更合适，因为它保证有初始值且总是持有最新值。

2. 我需要重复发射和收集同样的值吗？

   • 如果需要，那考虑 SharedFlow。因为 StateFlow 会忽略连续两次重复的值（基于 equals 比较）。

3. 当有新的订阅者订阅的时候，我需要发射最近的多个值吗？

   • 如果需要，那考虑 SharedFlow，可以配置 replay 参数。StateFlow 默认 replay 为 1。

SharedFlow 的高级配置

SharedFlow 允许配置 replay 和 onBufferOverflow：

• replay: 新订阅者能接收到的历史消息数量。
• onBufferOverflow: 当缓冲区满时的行为（如 DROP_OLDEST, CATCH, SUSPEND）。

JetPack Compose 中收集流的方式

关于在 UI 层收集 ViewModel 层的热流方式，官方文档已有介绍，但在 JetPack Compose 中推荐使用生命周期感知的方法。

先添加依赖：

implementation 'androidx.lifecycle:lifecycle-runtime-compose:2.6.0-alpha03'

代码示例

// 收集 StateFlow
val uiState by viewModel.userFlow.collectAsStateWithLifecycle()

// 收集 SharedFlow，区别在于需要赋初始值
val sharedState by viewModel.eventFlow.collectAsStateWithLifecycle(
    initialValue = UiState.Loading
)

when (uiState) {
    is UiState.Loading -> LoadingView()
    is UiState.Success -> SuccessView(uiState.data)
    is UiState.Error -> ErrorView(uiState.exception)
}

注意： 使用 collectAsStateWithLifecycle() 可以保证流的收集操作只发生在应用位于前台的时候，避免造成资源浪费。这与 collectAsState 不同，后者不会自动暂停/恢复。

单元测试实践

由于我们会在 ViewModel 中使用到 viewModelScope，首先可以定义一个 MainDispatcherRule，用于设置 MainDispatcher，确保协程在测试中同步执行。

1. 设置测试调度器

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.test.TestDispatcher
import kotlinx.coroutines.test.UnconfinedTestDispatcher
import kotlinx.coroutines.test.resetMain
import kotlinx.coroutines.test.setMain
import org.junit.rules.TestWatcher
import org.junit.runner.Description

class MainDispatcherRule(
  val testDispatcher: TestDispatcher = UnconfinedTestDispatcher()
) : TestWatcher() {
  override fun starting(description: Description) {
    super.starting(description)
    Dispatchers.setMain(testDispatcher)
  }

  override fun finished(description: Description) {
    super.finished(description)
    Dispatchers.resetMain()
  }
}

将规则应用到测试类中：

class MyViewModelTest {
  @get:Rule
  val mainDispatcherRule = MainDispatcherRule()
  
  // ... 测试方法
}

2. 测试 StateFlow

对于 StateFlow，可以直接访问 .value 属性来获取当前状态，无需等待收集完成。

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun `when initialized, repository emits loading and data`() = runTest {
    // arrange
    val viewModel = MyViewModel(repository)
    val users = listOf(User("Alice"), User("Bob"))
    
    // 初始值应该是 UiState.Loading，因为 stateFlow 可以直接获取最新值
    assertEquals(UiState.Loading, viewModel.userFlow.value)

    // action
    repository.sendUsers(users)
    viewModel.onRefresh()

    // check
    assertEquals(UiState.Success(users), viewModel.userFlow.value)
}

Mock Repository 实现：

class TestUserRepository : UserRepository {
  private val usersFlow = MutableSharedFlow<List<User>>(replay = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)

  override fun getUsers(): Flow<List<User>> = usersFlow

  suspend fun sendUsers(users: List<User>) {
    usersFlow.emit(users)
  }
}

使用 stateIn 时的测试： 当 ViewModel 内部使用 stateIn 时，外部无法直接通过 .value 捕获所有中间状态，需要在测试中启动收集。

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun `test stateIn with external collection`() = runTest {
    val viewModel = MainWithStateinViewModel(repository)
    val users = listOf(User("Alice"))
    
    // 启动收集以触发 stateIn 的逻辑
    val collectJob = launch(UnconfinedTestDispatcher(testScheduler)) {
        viewModel.userFlow.collect()
    }
    
    assertEquals(UiState.Loading, viewModel.userFlow.value)
    repository.sendUsers(users)
    assertEquals(UiState.Success(users), viewModel.userFlow.value)
    
    collectJob.cancel()
}

3. 测试 SharedFlow

测试 SharedFlow 可以使用 Turbine 库，它提供了更流畅的断言 API。

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun `test SharedFlow events`() = runTest {
    val viewModel = MainWithShareInViewModel(repository)
    val users = listOf(User("Alice"))
    
    repository.sendUsers(users)

    viewModel.userFlow.test {
        val firstItem = awaitItem()
        assertEquals(UiState.Loading, firstItem)

        val secondItem = awaitItem()
        assertEquals(UiState.Success(users), secondItem)
    }
}

Turbine 的 test 扩展函数会自动处理协程作用域，简化了测试代码。

最佳实践总结

1. 统一数据源：ViewModel 应作为 UI 的唯一可信数据源，避免 View 直接调用 Repository。
2. 生命周期安全：始终使用 viewModelScope 和 SharingStarted.WhileSubscribed 防止内存泄漏。
3. 状态不可变：尽量使用 Immutable 对象作为 StateFlow 的类型，避免意外修改。
4. 错误处理：在 asResult 等转换层统一处理异常，不要将原始 Exception 暴露给 UI。
5. 测试覆盖：针对冷流转热流、边界条件、取消操作编写单元测试，确保逻辑健壮。

通过合理使用 StateFlow 和 SharedFlow，并结合完善的单元测试，可以显著提升 Android 应用的稳定性和可维护性。