Kotlin协程之Flow基础原理示例解析

launch(Dispatchers.Main) {
    flow {
        emit(1)
        emit(2)
    }.collect {
        delay(1000)

        withContext(Dispatchers.IO) {
            Log.d("liduo", "$it")
        }

        Log.d("liduo", "$it")
    }
}

public fun  flow(@BuilderInference block: suspend FlowCollector.() -> Unit): Flow = SafeFlow(block)

public interface FlowCollector {
    // 可挂起，非线程安全
    public suspend fun emit(value: T)
}

public interface Flow {
   
    @InternalCoroutinesApi
    public suspend fun collect(collector: FlowCollector)
}

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

private class SafeFlow(private val block: suspend FlowCollector.() -> Unit) : AbstractFlow() {
    override suspend fun collectSafely(collector: FlowCollector) {
        collector.block()
    }
}

@FlowPreview
public abstract class AbstractFlow : Flow, CancellableFlow {

     // 核心方法 
    @InternalCoroutinesApi
    public final override suspend fun collect(collector: FlowCollector) {
        // 创建SafeCollector对象，对collector进行包裹
        val safeCollector = SafeCollector(collector, coroutineContext)
        try {
            // 调用collectSafely方法
            collectSafely(safeCollector)
        } finally {
            // 释放拦截的续体
            safeCollector.releaseIntercepted()
        }
    }
    
    public abstract suspend fun collectSafely(collector: FlowCollector)
}

internal actual class SafeCollector actual constructor(
    @JvmField internal actual val collector: FlowCollector,
    @JvmField internal actual val collectContext: CoroutineContext
) : FlowCollector, ContinuationImpl(NoOpContinuation, EmptyCoroutineContext), CoroutineStackFrame {
    
    ...
    // 保存上下文中元素数量，用于检查上下文是否变化
    @JvmField
    internal actual val collectContextSize = collectContext.fold(0) { count, _ -> count + 1 }
    // 保存上一次的上下文
    private var lastEmissionContext: CoroutineContext? = null
    // 执行结束后的续体
    private var completion: Continuation? = null

    // 协程上下文
    override val context: CoroutineContext
        get() = completion?.context ?: EmptyCoroutineContext

    // 挂起的核心方法
    override fun invokeSuspend(result: Result): Any? {
        result.onFailure { lastEmissionContext = DownstreamExceptionElement(it) }
        completion?.resumeWith(result as Result)
        return COROUTINE_SUSPENDED
    }

    // 释放拦截的续体
    public actual override fun releaseIntercepted() {
        super.releaseIntercepted()
    }

    // 发射数据
    override suspend fun emit(value: T) {
        // 获取当前suspend方法续体
        return suspendCoroutineUninterceptedOrReturn sc@{ uCont ->
            try {
                // 调用重载的方法
                emit(uCont, value)
            } catch (e: Throwable) {
                 // 出现异常时，将异常封装成上下文，保存到lastEmissionContext
                lastEmissionContext = DownstreamExceptionElement(e)
                // 抛出异常
                throw e
            }
        }
    }

    // 重载的emit方法
    private fun emit(uCont: Continuation, value: T): Any? {
        // 从续体中获取上下文
        val currentContext = uCont.context
        // 保证当前协程的Job是active的
        currentContext.ensureActive()
        // 获取上次的上下文
        val previousContext = lastEmissionContext
        // 如果前后上下文发生变化
        if (previousContext !== currentContext) {
            // 检查上下文是否发生异常
            checkContext(currentContext, previousContext, value)
        }
        // 保存续体
        completion = uCont
        // 调用emitFun方法，传入collector，value，continuation
        return emitFun(collector as FlowCollector, value, this as Continuation)
    }

    // 检查上下文变化，防止并发
    private fun checkContext(
        currentContext: CoroutineContext,
        previousContext: CoroutineContext?,
        value: T
    ) {
        // 如果上次执行过程中发生了异常
        if (previousContext is DownstreamExceptionElement) {
            // 抛出异常
            exceptionTransparencyViolated(previousContext, value)
        }
        // 检查上下文是否发生变化，如果变化，则抛出异常
        checkContext(currentContext)
        lastEmissionContext = currentContext
    }
    
    // 用于抛出异常
    private fun exceptionTransparencyViolated(exception: DownstreamExceptionElement, value: Any?) {
        error("""
            Flow exception transparency is violated:
                Previous 'emit' call has thrown exception ${exception.e}, but then emission attempt of value '$value' has been detected.
                Emissions from 'catch' blocks are prohibited in order to avoid unspecified behaviour, 'Flow.catch' operator can be used instead.
                For a more detailed explanation, please refer to Flow documentation.
            """.trimIndent())
    }
}

private val emitFun =
    FlowCollector::emit as Function3, Any?, Continuation, Any?>

@Nullable
public final Object invoke(@NotNull FlowCollector p1, @Nullable Object p2, @NotNull Continuation continuation) {
   InlineMarker.mark(0);
   // 核心执行
   Object var10000 = p1.emit(p2, continuation);
   InlineMarker.mark(2);
   InlineMarker.mark(1);
   return var10000;
}

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

override suspend fun emit(value: T) {
    // 获取当前suspend方法续体
    return suspendCoroutineUninterceptedOrReturn sc@{ uCont ->
        try {
            // 调用重载的方法
            emit(uCont, value)
        } catch (e: Throwable) {
            // 出现异常时，将异常封装成上下文，保存到lastEmissionContext
            lastEmissionContext = DownstreamExceptionElement(e)
            // 抛出异常
            throw e
        }
    }
}

emitFun(collector as FlowCollector, value, this as Continuation)

override fun invokeSuspend(result: Result): Any? {
    // 尝试获取异常
    result.onFailure { lastEmissionContext = DownstreamExceptionElement(it) }
    // 如果没有异常，则恢复flow方法续体的执行
    completion?.resumeWith(result as Result)
    // 返回挂起标识，这里挂起的是消费过程
    return COROUTINE_SUSPENDED
}