ViewModel 中 StateFlow 与 SharedFlow 的选择策略及单元测试实践

shareIn 拓展函数使用方式也是类似的，只不过没有初始值 initialValue 参数，且默认行为是立即开始共享，需配合 replay 参数控制重放数量。

StateFlow 与 SharedFlow 如何选择？

上文说到，我们应该在 ViewModel 中暴露出热流，现在我们有两个热流-StateFlow 和 SharedFlow，如何选择？没什么特定的规则，选择的时候只需要想一下以下几个问题：

1. 我真的需要在特定的时间、位置获取 Flow 的最新状态吗？ 如果不需要，那考虑 SharedFlow，比如常用的事件通知功能（如 Toast 提示、导航跳转）。

2. 我需要重复发射和收集同样的值吗？ 如果需要，那考虑 SharedFlow，因为 StateFlow 会忽略连续两次重复的值（基于 equals 比较），而 SharedFlow 可以配置重放策略。

3. 当有新的订阅者订阅的时候，我需要发射最近的多个值吗？ 如果需要，那考虑 SharedFlow，可以配置 replay 参数。StateFlow 只能重放最后一个值。

4. 是否需要初始值？ StateFlow 必须有初始值，SharedFlow 可选。UI 状态通常需要初始值来渲染加载态。

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 uiState by viewModel.userFlow.collectAsStateWithLifecycle(
    initialValue = UiState.Loading
)

when(uiState) {
    is UiState.Loading -> {
        // 显示加载动画
    }
    is UiState.Success -> {
        // 显示数据
    }
    is UiState.Error -> {
        // 显示错误信息
    }
}

使用 collectAsStateWithLifecycle() 也是可以保证流的收集操作只发生在应用位于前台的时候，避免造成资源浪费。这比普通的 collectAsState() 更安全，因为它会自动处理 Lifecycle 状态。

单元测试实践

由于我们会在 ViewModel 中使用到 viewModelScope，首先可以定义一个 MainDispatcherRule，用于设置 MainDispatcher，确保测试在单线程下运行，便于调试。

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.TestRule
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()
  }
}

将 MainDispatcherRule 用于 ViewModel 单元测试代码中：

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

1. 测试 StateFlow

现在我们有一个业务 ViewModel 如下：

@HiltViewModel
class MyViewModel @Inject constructor(
  private val userRepository: UserRepository
) : ViewModel() {

  private val _userFlow = MutableStateFlow<UiState>(UiState.Loading)
  val userFlow: StateFlow<UiState> = _userFlow.asStateFlow()

  fun onRefresh() {
    viewModelScope.launch {
      userRepository
        .getUsers().asResult()
        .collect { result ->
          _userFlow.update {
            when (result) {
              is Result.Loading -> UiState.Loading
              is Result.Success -> UiState.Success(result.data)
              is Result.Error -> UiState.Error(result.exception)
            }
          }
        }
    }
  }
}

单元测试代码如下：

class MyViewModelTest{
    @get:Rule
    val mainDispatcherRule = MainDispatcherRule()

    // arrange
    private val repository = TestUserRepository()

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

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

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

// Mock UserRepository
class TestUserRepository : UserRepository {

  /**
   * The backing hot flow for the list of users for testing.
   */
  private val usersFlow =
    MutableSharedFlow<List<User>>(replay = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)

  override fun getUsers(): Flow<List<User>> {
    return usersFlow
  }

  /**
   * A test-only API to allow controlling the list of users from tests.
   */
  suspend fun sendUsers(users: List<User>) {
    usersFlow.emit(users)
  }
}

如果 ViewModel 中使用的是 stateIn 拓展函数：

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun `when initialized, repository emits loading and data`() = runTest {
    //arrange
    val viewModel = MainWithStateinViewModel(repository)
    val users = listOf(User("id1"))
    
    //action
    // 因为此时 collect 操作并不是在 ViewModel 中，我们需要在测试代码中执行 collect
    val collectJob = launch(UnconfinedTestDispatcher(testScheduler)) {
        viewModel.userFlow.collect()
    }
    
    //check
    assertEquals(UiState.Loading, viewModel.userFlow.value)
    
    //action
    repository.sendUsers(users)
    
    //check
    assertEquals(UiState.Success(users), viewModel.userFlow.value)

    collectJob.cancel()
}

2. 测试 SharedFlow

测试 SharedFlow 可以使用一个开源库，Turbine 是一个用于测试 Flow 的小型开源库，它能提供更清晰的断言接口。

测试使用 sharedIn 拓展函数的 SharedFlow：

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun `when initialized, repository emits loading and data`() = runTest {
    val viewModel = MainWithShareInViewModel(repository)

    val users = listOf(User("id1"))
    repository.sendUsers(users)

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

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

使用 Turbine 的 .test {} 块可以自动管理协程作用域，并且提供 awaitItem()、awaitComplete()、awaitError() 等方法来验证流的行为。

最佳实践与常见陷阱

1. 内存泄漏：确保在 ViewModel 销毁时取消所有协程。使用 viewModelScope 可以自动处理，但手动启动协程时需小心。
2. 生命周期感知：在 Compose 中优先使用 collectAsStateWithLifecycle()，避免在非前台状态下消耗资源。
3. 并发控制：在测试中务必使用 runTest 和自定义 TestDispatcher，否则异步操作可能导致测试不稳定。
4. 错误处理：确保 Flow 中的异常被正确捕获并转换为 UiState.Error，避免应用崩溃。
5. 初始值：StateFlow 必须提供初始值，防止 UI 在首次加载前处于空白状态。