launch(Dispatchers.Main) {
  flow {
    emit(1)
    emit(2)
  }.flowOn(Dispatchers.IO).collect {
    delay(1000)
    withContext(Dispatchers.IO) {
      Log.d("liduo", "$it")
    }
    Log.d("liduo", "$it")
  }
}

launch(Dispatchers.Main) {
  flow {
    emit(2) // 執(zhí)行在IO線程池
  }.flowOn(Dispatchers.IO).map {
    it + 1 // 執(zhí)行在Default線程池
  }.flowOn(Dispatchers.Default).collect {
    Log.d("liduo", "$it") //執(zhí)行在主線程
  }
}

public fun <T> Flow<T>.flowOn(context: CoroutineContext): Flow<T> {
    // 檢查當(dāng)前協(xié)程沒(méi)有執(zhí)行結(jié)束
    checkFlowContext(context)
    return when {
        // 為空，則返回自身
        context == EmptyCoroutineContext -> this
        // 如果是可融合的Flow，則嘗試融合操作，獲取新的流
        this is FusibleFlow -> fuse(context = context)
        // 其他情況，包裝成可融合的Flow
        else -> ChannelFlowOperatorImpl(this, context = context)
    }
}
// 確保Job不為空
private fun checkFlowContext(context: CoroutineContext) {
    require(context[Job] == null) {
        "Flow context cannot contain job in it. Had $context"
    }
}

internal class ChannelFlowOperatorImpl<T>(
    flow: Flow<T>,
    context: CoroutineContext = EmptyCoroutineContext,
    capacity: Int = Channel.OPTIONAL_CHANNEL,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
) : ChannelFlowOperator<T, T>(flow, context, capacity, onBufferOverflow) {
    // 用于流融合時(shí)創(chuàng)建新的流
    override fun create(context: CoroutineContext, capacity: Int, onBufferOverflow: BufferOverflow): ChannelFlow<T> =
        ChannelFlowOperatorImpl(flow, context, capacity, onBufferOverflow)
    // 若當(dāng)前的流不需要通過(guò)Channel即可實(shí)現(xiàn)正常工作時(shí)，會(huì)調(diào)用此方法
    override fun dropChannelOperators(): Flow<T>? = flow
    // 觸發(fā)對(duì)下一級(jí)流進(jìn)行收集
    override suspend fun flowCollect(collector: FlowCollector<T>) =
        flow.collect(collector)
}

public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value: T) -> Unit): Unit = collect(object : FlowCollector<T> {<!--{C}%3C!%2D%2D%20%2D%2D%3E--> override suspend fun emit(value: T) = action(value) })public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value: T) -> Unit): Unit =
    collect(object : FlowCollector<T> {
        override suspend fun emit(value: T) = action(value)
    })

override suspend fun collect(collector: FlowCollector<T>) {
    // OPTIONAL_CHANNEL為默認(rèn)值，這里滿足條件，之后會(huì)詳細(xì)講解
    if (capacity == Channel.OPTIONAL_CHANNEL) {
        // 獲取當(dāng)前協(xié)程的上下文
        val collectContext = coroutineContext
        // 計(jì)算新的上下文
        val newContext = collectContext + context
        // 如果前后上下文沒(méi)有發(fā)生變化
        if (newContext == collectContext)
            // 直接觸發(fā)對(duì)下一級(jí)流的收集
            return flowCollect(collector)
        // 如果上下文發(fā)生變化，但不需要切換線程
        if (newContext[ContinuationInterceptor] == collectContext[ContinuationInterceptor])
            // 切換協(xié)程上下文，調(diào)用flowCollect方法觸發(fā)下一級(jí)流的收集
            return collectWithContextUndispatched(collector, newContext)
    }
    // 調(diào)用父類(lèi)的collect方法
    super.collect(collector)
}
// 獲取當(dāng)前協(xié)程的上下文，該方法會(huì)被編譯器處理
@SinceKotlin("1.3")
@Suppress("WRONG_MODIFIER_TARGET")
@InlineOnly
public suspend inline val coroutineContext: CoroutineContext
    get() {
        throw NotImplementedError("Implemented as intrinsic")
    }

override suspend fun collect(collector: FlowCollector<T>): Unit =
    coroutineScope {
        collector.emitAll(produceImpl(this))
    }

public open fun produceImpl(scope: CoroutineScope): ReceiveChannel<T> =
    scope.produce(context, produceCapacity, onBufferOverflow, start = CoroutineStart.ATOMIC, block = collectToFun)

internal fun <E> CoroutineScope.produce(
    context: CoroutineContext = EmptyCoroutineContext,
    capacity: Int = 0,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    onCompletion: CompletionHandler? = null,
    @BuilderInference block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E> {
    // 根據(jù)容量與溢出策略創(chuàng)建Channel對(duì)象
    val channel = Channel<E>(capacity, onBufferOverflow)
    // 計(jì)算新的上下文
    val newContext = newCoroutineContext(context)
    // 創(chuàng)建協(xié)程
    val coroutine = ProducerCoroutine(newContext, channel)
    // 監(jiān)聽(tīng)完成事件
    if (onCompletion != null) coroutine.invokeOnCompletion(handler = onCompletion)
    // 啟動(dòng)協(xié)程
    coroutine.start(start, coroutine, block)
    return coroutine
}

internal val collectToFun: suspend (ProducerScope<T>) -> Unit
    get() = { collectTo(it) }

protected override suspend fun collectTo(scope: ProducerScope<T>) =
    flowCollect(SendingCollector(scope))

override suspend fun flowCollect(collector: FlowCollector<T>) =
    flow.collect(collector)

private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
    override suspend fun collectSafely(collector: FlowCollector<T>) {
        // 觸發(fā)執(zhí)行
        collector.block()
    }
}

@InternalCoroutinesApi
public class SendingCollector<T>(
    private val channel: SendChannel<T>
) : FlowCollector<T> {
    // 通過(guò)Channel類(lèi)對(duì)象發(fā)送值
    override suspend fun emit(value: T): Unit = channel.send(value)
}

override suspend fun collect(collector: FlowCollector<T>): Unit =
    coroutineScope {
        collector.emitAll(produceImpl(this))
    }

public suspend fun <T> FlowCollector<T>.emitAll(channel: ReceiveChannel<T>): Unit =
    emitAllImpl(channel, consume = true)

private suspend fun <T> FlowCollector<T>.emitAllImpl(channel: ReceiveChannel<T>, consume: Boolean) {
    // 用于保存異常
    var cause: Throwable? = null
    try {
        // 死循環(huán)
        while (true) {
            // 掛起，等待接收Channel結(jié)果或Channel關(guān)閉
            val result = run { channel.receiveOrClosed() }
            // 如果Channel關(guān)閉了
            if (result.isClosed) {
                // 如果有異常，則拋出
                result.closeCause?.let { throw it }
                // 沒(méi)有異常，則跳出循環(huán)
                break
            }
            // 獲取并發(fā)送值
            emit(result.value)
        }
    } catch (e: Throwable) {
        // 捕獲到異常時(shí)拋出
        cause = e
        throw e
    } finally {
         // 執(zhí)行結(jié)束關(guān)閉Channel
        if (consume) channel.cancelConsumed(cause)
    }
}

public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value: T) -> Unit): Unit =
    collect(object : FlowCollector<T> {
        override suspend fun emit(value: T) = action(value)
    })

launch(Dispatchers.Main) {
    flow {
        emit(2)
      // emit方法是在IO線程執(zhí)行還是在主線程執(zhí)行呢？  
    }.flowOn(Dispatchers.IO).flowOn(Dispatchers.Main).collect {
        Log.d("liduo", "$it")
    }
}

public fun <T> Flow<T>.flowOn(context: CoroutineContext): Flow<T> {
    // 檢查當(dāng)前協(xié)程沒(méi)有執(zhí)行結(jié)束
    checkFlowContext(context)
    return when {
        // 為空，則返回自身
        context == EmptyCoroutineContext -> this
        // 如果是可融合的Flow，則嘗試融合操作，獲取新的流
        this is FusibleFlow -> fuse(context = context)
        // 其他情況，包裝成可融合的Flow
        else -> ChannelFlowOperatorImpl(this, context = context)
    }
}

public override fun fuse(context: CoroutineContext, capacity: Int, onBufferOverflow: BufferOverflow): Flow<T> {
    ...
    // 計(jì)算融合后流的上下文
    // context為下游的上下文，this.context為上游的上下文 
    val newContext = context + this.context
    ...
}

public operator fun plus(other: CoroutineDispatcher): CoroutineDispatcher = other