第九章：协程与Flow测试——TestCoroutineDispatcher详解

协程和Flow，现在已经是Android开发的标配了。但说实话，我刚接触协程测试那会儿，也踩过不少坑。尤其是TestCoroutineDispatcher，名字听着挺唬人，其实用起来就那么几招。今天咱们就把这块彻底聊透。

9.1 TestCoroutineDispatcher详解

先说说这个调度器是干嘛的。简单讲，它让你能手动控制协程的执行时机。你想想看，生产环境里协程是异步的，但测试里我们希望它同步——这样才好断言结果。

核心要点：TestCoroutineDispatcher 替代了原来的 TestCoroutineScope，是官方推荐的测试调度器。

// 老方式（已废弃）
val testScope = TestCoroutineScope()

// 新方式（推荐）
val testDispatcher = StandardTestDispatcher()
val testScope = TestScope(testDispatcher)

我个人习惯用TestScope，因为它自带advanceUntilIdle()和runCurrent()，测试起来特别顺手。

9.1.1 手动控制时间

这里有个我经常用的技巧：用advanceTimeBy()来模拟延迟。

@Test
fun `测试延迟发射`() = runTest {
    val dispatcher = StandardTestDispatcher()
    
    var result = ""
    
    // 启动一个协程，延迟1秒后赋值
    backgroundScope.launch(dispatcher) {
        delay(1000)
        result = "完成"
    }
    
    // 此时result还是空的
    assertEquals("", result)
    
    // 手动推进1秒
    dispatcher.scheduler.advanceTimeBy(1000)
    dispatcher.scheduler.runCurrent()
    
    // 现在result有值了
    assertEquals("完成", result)
}

嗯，这里要注意：advanceTimeBy只是推进时间，不会自动执行挂起点后的代码。你得再调一次runCurrent()才能触发执行。我曾经在这上面浪费过半小时排查为什么断言不通过……

9.2 测试协程作用域

协程作用域的测试，说白了就是验证你的协程有没有按预期启动、取消、或者泄漏。

9.2.1 验证协程启动

class MyViewModel(
    private val scope: CoroutineScope = CoroutineScope(Dispatchers.Main)
) {
    private val _state = MutableStateFlow("初始")
    val state: StateFlow<String> = _state
    
    fun loadData() {
        scope.launch {
            delay(100)
            _state.value = "已加载"
        }
    }
}

@Test
fun `验证协程启动后状态变化`() = runTest {
    val testDispatcher = StandardTestDispatcher()
    val scope = TestScope(testDispatcher)
    val viewModel = MyViewModel(scope)
    
    viewModel.loadData()
    
    // 推进时间，让协程执行
    testDispatcher.scheduler.advanceUntilIdle()
    
    assertEquals("已加载", viewModel.state.value)
}

小技巧：用advanceUntilIdle()比手动advanceTimeBy更省心。它会一直推进到所有待执行协程都跑完。

9.2.2 验证协程取消

我记得有一次线上出了个bug，协程在页面销毁后还在跑，导致内存泄漏。从那以后，我特别重视协程取消的测试。

@Test
fun `验证协程取消后不再执行`() = runTest {
    val testDispatcher = StandardTestDispatcher()
    val scope = TestScope(testDispatcher)
    
    var executed = false
    
    val job = scope.launch {
        delay(1000)
        executed = true
    }
    
    // 取消协程
    job.cancel()
    
    // 推进时间
    testDispatcher.scheduler.advanceTimeBy(2000)
    testDispatcher.scheduler.runCurrent()
    
    // 断言：协程被取消，executed应该还是false
    assertFalse(executed)
}

注意：取消协程后，advanceUntilIdle()不会执行被取消的协程。但如果你用advanceTimeBy手动推进，要确保调用了runCurrent()。

9.3 测试Flow发射与收集

Flow的测试，核心就两个动作：发射和收集。咱们一个个来看。

9.3.1 测试Flow发射

fun produceNumbers(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}

@Test
fun `验证Flow按顺序发射`() = runTest {
    val testDispatcher = StandardTestDispatcher()
    
    val results = mutableListOf<Int>()
    
    // 收集Flow
    val job = launch(testDispatcher) {
        produceNumbers().collect { results.add(it) }
    }
    
    // 推进到所有发射完成
    testDispatcher.scheduler.advanceUntilIdle()
    
    assertEquals(listOf(1, 2, 3), results)
    
    job.cancel()
}

你可能会问：为什么不用flow.take(3).toList()？嗯，那样确实更简洁。但真实项目中，Flow往往有副作用，用collect能更真实地模拟生产环境。

9.3.2 使用Turbine测试Flow

我个人特别喜欢用Turbine这个库，它让Flow测试变得特别优雅。

// build.gradle.kts
// testImplementation("app.cash.turbine:turbine:1.0.0")

@Test
fun `使用Turbine测试Flow`() = runTest {
    val flow = flow {
        emit(1)
        emit(2)
        emit(3)
    }
    
    flow.test {
        assertEquals(1, awaitItem())
        assertEquals(2, awaitItem())
        assertEquals(3, awaitItem())
        awaitComplete()
    }
}

为什么推荐Turbine？因为它自动处理了协程调度，你不需要手动advanceUntilIdle。而且它的错误信息特别清晰，哪个断言失败了一目了然。

9.4 测试Flow的异常与取消

Flow的异常处理，是很多开发者容易忽略的地方。我见过太多线上崩溃，就是因为Flow抛异常没被捕获。

9.4.1 测试Flow异常

fun failingFlow(): Flow<Int> = flow {
    emit(1)
    throw RuntimeException("出错了")
    emit(2) // 这行不会执行
}

@Test
fun `验证Flow抛出异常`() = runTest {
    val results = mutableListOf<Int>()
    
    try {
        failingFlow().collect { results.add(it) }
    } catch (e: RuntimeException) {
        assertEquals("出错了", e.message)
    }
    
    assertEquals(listOf(1), results) // 只有第一个元素被发射
}

用Turbine就更简单了：

@Test
fun `Turbine验证Flow异常`() = runTest {
    failingFlow().test {
        assertEquals(1, awaitItem())
        awaitError() // 验证抛出了异常
    }
}

9.4.2 测试Flow取消

我曾经在项目里遇到过一个坑：Flow在后台一直跑，页面销毁了还在发射数据，导致UI更新时崩溃。所以，验证Flow能否被正确取消，真的很重要。

@Test
fun `验证Flow在取消后停止发射`() = runTest {
    val testDispatcher = StandardTestDispatcher()
    val scope = TestScope(testDispatcher)
    
    var emissionCount = 0
    
    val job = scope.launch {
        flow {
            while (true) {
                delay(100)
                emit(++emissionCount)
            }
        }.collect { /* 什么都不做 */ }
    }
    
    // 推进200毫秒，应该发射了2次
    testDispatcher.scheduler.advanceTimeBy(200)
    testDispatcher.scheduler.runCurrent()
    assertEquals(2, emissionCount)
    
    // 取消协程
    job.cancel()
    
    // 再推进200毫秒，发射次数不应该增加
    testDispatcher.scheduler.advanceTimeBy(200)
    testDispatcher.scheduler.runCurrent()
    assertEquals(2, emissionCount) // 还是2，说明Flow被取消了
}

9.5 测试Channel

Channel的测试，比Flow稍微复杂一点。因为它涉及到发送和接收的同步问题。

9.5.1 基本Channel测试

@Test
fun `测试Channel发送和接收`() = runTest {
    val testDispatcher = StandardTestDispatcher()
    val scope = TestScope(testDispatcher)
    val channel = Channel<Int>(capacity = Channel.UNLIMITED)
    
    // 发送数据
    scope.launch {
        channel.send(1)
        channel.send(2)
        channel.send(3)
        channel.close()
    }
    
    // 接收数据
    val results = mutableListOf<Int>()
    scope.launch {
        for (item in channel) {
            results.add(item)
        }
    }
    
    testDispatcher.scheduler.advanceUntilIdle()
    
    assertEquals(listOf(1, 2, 3), results)
}

9.5.2 测试Channel的背压

这里有个我踩过的坑：用Channel.RENDEZVOUS时，发送和接收必须配对，否则会挂起。

@Test
fun `测试RENDEZVOUS Channel的背压`() = runTest {
    val testDispatcher = StandardTestDispatcher()
    val scope = TestScope(testDispatcher)
    val channel = Channel<Int>(capacity = Channel.RENDEZVOUS)
    
    // 发送数据（会挂起，直到有人接收）
    scope.launch {
        channel.send(1)
        channel.send(2)
    }
    
    // 接收数据
    scope.launch {
        assertEquals(1, channel.receive())
        assertEquals(2, channel.receive())
    }
    
    testDispatcher.scheduler.advanceUntilIdle()
}

注意：如果发送和接收不配对，测试会一直挂起。记得设置超时时间，或者用withTimeout包裹。

9.5.3 测试Channel的取消

@Test
fun `测试Channel取消后不再接收`() = runTest {
    val testDispatcher = StandardTestDispatcher()
    val scope = TestScope(testDispatcher)
    val channel = Channel<Int>(capacity = Channel.UNLIMITED)
    
    channel.send(1)
    channel.send(2)
    
    val job = scope.launch {
        channel.consumeEach { /* 消费 */ }
    }
    
    job.cancel()
    
    // 取消后，channel应该已经关闭
    assertTrue(channel.isClosedForReceive)
}

知识体系总览

说了这么多，咱们用一张图来总结一下本章的核心内容：

协程与Flow测试知识体系 TestCoroutineDispatcher 手动控制协程执行时机 advanceTimeBy / runCurrent advanceUntilIdle 测试协程作用域 验证协程启动 验证协程取消 防止协程泄漏 Flow发射与收集 collect / toList Turbine库：awaitItem 验证发射顺序 Flow异常与取消 try-catch捕获异常 Turbine: awaitError 测试Channel send / receive 配对 背压与取消测试

这张图把本章的五个核心知识点串起来了。从TestCoroutineDispatcher这个基础工具出发，延伸到协程作用域、Flow和Channel的测试。每个模块都有对应的测试方法和工具。

我的建议：刚开始学的时候，先掌握TestCoroutineDispatcher和runTest这两个基础。等用熟了，再慢慢引入Turbine这样的第三方库。别一口气吃成胖子，测试这东西，稳扎稳打最重要。

好了，这一章的内容就到这儿。记住，测试协程和Flow的核心就四个字：控制时间。只要你能控制住协程的执行时机，剩下的就是写断言的事了。

公众号:蓝海资料掘金营，微信deep3321

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