kotlin - onEach 更改 StateFlow 中的调度程序(kotlin 协程)

Question

想象一下下面的自包含测试用例

@Test
fun `stateFlow in GlobalScope`() = runBlockingTest {

    suspend fun makeHeavyRequest(): String {
        return "heavy result"
    }

    val flow1 = flowOf(Unit)
        .map { makeHeavyRequest() }
        .onEach { logThread("1: before flowOn") }
        .flowOn(testDispatcher)
        .stateIn(GlobalScope, SharingStarted.Lazily, "init state")

    val flow2 = flowOf(Unit)
        .map { makeHeavyRequest() }
        .onEach { logThread("2: before flowOn") }
        .flowOn(testDispatcher)
        .stateIn(GlobalScope, SharingStarted.Lazily, "init state")
        .onEach { logThread("2: after stateIn") }

    val flow3 = flowOf(Unit)
        .map { makeHeavyRequest() }
        .onEach { logThread("3: before flowOn") }
        .flowOn(testDispatcher)
        .onEach { logThread("3: after flowOn") }
        .stateIn(GlobalScope, SharingStarted.Lazily, "init state")

    flow1.test {
        assertEquals("heavy result", expectItem())
        cancelAndIgnoreRemainingEvents()
    }

    flow2.test {
        assertEquals("heavy result", expectItem())
        cancelAndIgnoreRemainingEvents()
    }

    flow3.test {
        assertEquals("heavy result", expectItem())
        cancelAndIgnoreRemainingEvents()
    }

}

运行效果会是:

Thread (1: before flowOn): Thread[main @coroutine#2,5,main]
Thread (2: before flowOn): Thread[main @coroutine#3,5,main]
Thread (2: after stateIn): Thread[main @coroutine#6,5,main]
Thread (3: before flowOn): Thread[DefaultDispatcher-worker-1 @coroutine#8,5,main]
Thread (3: after flowOn): Thread[DefaultDispatcher-worker-1 @coroutine#4,5,main]


org.opentest4j.AssertionFailedError: 
Expected :heavy result
Actual   :init state

在 flow3 中，将 onEach 放在 flowOn 和 stateIn 之间完全改变了调度程序并搞砸了结果。这是为什么？

Answer 1

发生这种情况的原因是因为 stateIn 运算符根据上游流是否为 ChannelFlow 进行了一些优化。
.flowOn(...) 返回一个 ChannelFlow 而 .onEach(...) 没有。

通常这并不重要。为什么它对您的情况很重要，是因为您希望 stateIn 返回的流永远不会发出初始值。但是这个参数是强制性的是有原因的，你应该期望收到初始值。您是否真的这样做主要取决于上游流是否能够在不挂起的情况下发出值。

现在看来，stateIn 运算符中的优化之一是，它可能会在不挂起的情况下消耗 ChannelFlow。这就是为什么您在使用时会得到预期的行为

.flowOn(testDispatcher) /*returns ChannelFlow*/
.stateIn(GlobalScope, SharingStarted.Lazily, "init state")