图解Channel：如何理解它的CSP通信模型？

Channel 背后的数据结构

// 代码段1

fun main()  {
    val scope = CoroutineScope(Job() + mySingleDispatcher)
    // 1，创建管道
    val channel = Channel<Int>()

    scope.launch {
        // 2，在一个单独的协程当中发送管道消息
        repeat(3)  {
            channel.send(it)
            println("Send: $it")
        }

        channel.close()
    }

    scope.launch {
        // 3，在一个单独的协程当中接收管道消息
        repeat(3) {
            val result = channel.receive()
            println("Receive ${result}")
        }
    }

    println("end")
    Thread.sleep(2000000L)
}

/*
输出结果：
end
Receive 0
Send: 0
Send: 1
Receive 1
Receive 2
Send: 2
*/

// 代码段2

public interface Channel<E> : SendChannel<E>, ReceiveChannel<E> {

public fun <E> Channel(
    capacity: Int = RENDEZVOUS,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    onUndeliveredElement: ((E) -> Unit)? = null
): Channel<E> =
    when (capacity) {
        RENDEZVOUS -> {
            if (onBufferOverflow == BufferOverflow.SUSPEND)
                RendezvousChannel(onUndeliveredElement)
            else
                ArrayChannel(1, onBufferOverflow, onUndeliveredElement)
        }
        CONFLATED -> {
            ConflatedChannel(onUndeliveredElement)
        }
        UNLIMITED -> LinkedListChannel(onUndeliveredElement)
        BUFFERED -> ArrayChannel(
            if (onBufferOverflow == BufferOverflow.SUSPEND) CHANNEL_DEFAULT_CAPACITY else 1,
            onBufferOverflow, onUndeliveredElement
        )
        else -> {
            if (capacity == 1 && onBufferOverflow == BufferOverflow.DROP_OLDEST)
                ConflatedChannel(onUndeliveredElement)
            else
                ArrayChannel(capacity, onBufferOverflow, onUndeliveredElement)
        }
    }

// 代码段3

internal abstract class AbstractSendChannel<E>(
    @JvmField protected val onUndeliveredElement: OnUndeliveredElement<E>?
) : SendChannel<E> {

    protected val queue = LockFreeLinkedListHead()

    // 省略

    internal abstract class AbstractChannel<E>(
    onUndeliveredElement: OnUndeliveredElement<E>?
) : AbstractSendChannel<E>(onUndeliveredElement), Channel<E> {}
}

// 代码段4

public actual open class LockFreeLinkedListHead : LockFreeLinkedListNode() {
    public actual val isEmpty: Boolean get() = next === this
}

public actual open class LockFreeLinkedListNode {
    // 1
    private val _next = atomic<Any>(this)
    private val _prev = atomic(this)
    private val _removedRef = atomic<Removed?>(null)
}

发送和接收的流程

// 代码段5

public final override suspend fun send(element: E) {
    // 1
    if (offerInternal(element) === OFFER_SUCCESS) return
    // 2
    return sendSuspend(element)
}

protected open fun offerInternal(element: E): Any {
    while (true) {
        // 3
        val receive = takeFirstReceiveOrPeekClosed() ?: return OFFER_FAILE
        // 省略
    }
}

private suspend fun sendSuspend(element: E): Unit = suspendCancellableCoroutineReusable sc@ { cont ->
    loop@ while (true) {
        if (isFullImpl) {
            // 4
            val send = if (onUndeliveredElement == null)
                SendElement(element, cont) else
                SendElementWithUndeliveredHandler(element, cont, onUndeliveredElement)
            val enqueueResult = enqueueSend(send)
            when {
                enqueueResult == null -> {
                    // 5
                    cont.removeOnCancellation(send)
                    return@sc
                }
                enqueueResult is Closed<*> -> {
                }
                enqueueResult === ENQUEUE_FAILED -> {}
                enqueueResult is Receive<*> -> {}
                else -> error("enqueueSend returned $enqueueResult")
            }
        }
        // 省略
    }
}

// 代码段6

public final override suspend fun receive(): E {
    // 1
    val result = pollInternal()

    @Suppress("UNCHECKED_CAST")
    if (result !== POLL_FAILED && result !is Closed<*>) return result as E
    // 2
    return receiveSuspend(RECEIVE_THROWS_ON_CLOSE)
}

protected open fun pollInternal(): Any? {
    while (true) {
        // 3
        val send = takeFirstSendOrPeekClosed() ?: return POLL_FAILED
        val token = send.tryResumeSend(null)
        if (token != null) {
            assert { token === RESUME_TOKEN }
            //4
            send.completeResumeSend()
            return send.pollResult
        }

        send.undeliveredElement()
    }
}

// CancellableContinuationImpl
private fun dispatchResume(mode: Int) {
    if (tryResume()) return
    // 5
    dispatch(mode)
}

internal fun <T> DispatchedTask<T>.dispatch(mode: Int) {
    // 省略
    if (!undispatched && delegate is DispatchedContinuation<*> && mode.isCancellableMode == resumeMode.isCancellableMode) {

        val dispatcher = delegate.dispatcher
        val context = delegate.context
        if (dispatcher.isDispatchNeeded(context)) {
            // 6
            dispatcher.dispatch(context, this)
        } else {
            resumeUnconfined()
        }
    } else {
        // 省略
    }
}

private suspend fun <R> receiveSuspend(receiveMode: Int): R = suspendCancellableCoroutineReusable sc@ { cont ->
    val receive = if (onUndeliveredElement == null)
        ReceiveElement(cont as CancellableContinuation<Any?>, receiveMode) else
        ReceiveElementWithUndeliveredHandler(cont as CancellableContinuation<Any?>, receiveMode, onUndeliveredElement)
    while (true) {
        if (enqueueReceive(receive)) {
            removeReceiveOnCancel(cont, receive)
            return@sc
        }

        val result = pollInternal()
        if (result is Closed<*>) {
            receive.resumeReceiveClosed(result)
            return@sc
        }
        if (result !== POLL_FAILED) {
            cont.resume(receive.resumeValue(result as E), receive.resumeOnCancellationFun(result as E))
            return@sc
        }
    }
}

小结

思考题

internal open class LinkedListChannel<E>(onUndeliveredElement: OnUndeliveredElement<E>?) : AbstractChannel<E>(onUndeliveredElement) {
    protected final override val isBufferAlwaysEmpty: Boolean get() = true
    protected final override val isBufferEmpty: Boolean get() = true
    protected final override val isBufferAlwaysFull: Boolean get() = false
    protected final override val isBufferFull: Boolean get() = false

    protected override fun offerInternal(element: E): Any {
        while (true) {
            val result = super.offerInternal(element)
            when {
                result === OFFER_SUCCESS -> return OFFER_SUCCESS
                result === OFFER_FAILED -> { // try to buffer
                    when (val sendResult = sendBuffered(element)) {
                        null -> return OFFER_SUCCESS
                        is Closed<*> -> return sendResult
                    }
                    // otherwise there was receiver in queue, retry super.offerInternal
                }
                result is Closed<*> -> return result
                else -> error("Invalid offerInternal result $result")
            }
        }
    }

    protected override fun offerSelectInternal(element: E, select: SelectInstance<*>): Any {
        while (true) {
            val result = if (hasReceiveOrClosed)
                super.offerSelectInternal(element, select) else
                (select.performAtomicTrySelect(describeSendBuffered(element)) ?: OFFER_SUCCESS)
            when {
                result === ALREADY_SELECTED -> return ALREADY_SELECTED
                result === OFFER_SUCCESS -> return OFFER_SUCCESS
                result === OFFER_FAILED -> {} // retry
                result === RETRY_ATOMIC -> {} // retry
                result is Closed<*> -> return result
                else -> error("Invalid result $result")
            }
        }
    }
}