Kotlin协程Channel源码示例浅析

fun main() = runBlocking {
    logX("开始")
    val channel = Channel<Int> {  }
    launch {
        (1..3).forEach{
            channel.send(it)
            logX("发送数据: $it")
        }
        // 关闭channel， 节省资源
        channel.close()
    }
    launch {
        for (i in channel){
            logX("接收数据: $i")
        }
    }
    logX("结束")
}

public fun <E> Channel(
    capacity: Int = RENDEZVOUS,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    onUndeliveredElement: ((E) -> Unit)? = null
): Channel<E> =
    when (capacity) {
        RENDEZVOUS -> {}
        CONFLATED -> {}
        UNLIMITED -> {}
        else -> {}
    }

public enum class BufferOverflow {
    /**
     * Suspend on buffer overflow.
     */
    SUSPEND,
    /**
     * Drop **the oldest** value in the buffer on overflow, add the new value to the buffer, do not suspend.
     */
    DROP_OLDEST,
    /**
     * Drop **the latest** value that is being added to the buffer right now on buffer overflow
     * (so that buffer contents stay the same), do not suspend.
     */
    DROP_LATEST
}

fun main() = runBlocking {
    println("开始")
    val channel = Channel<Int>(capacity = 2, onBufferOverflow = BufferOverflow.DROP_OLDEST) {
        println("onUndeliveredElement = $it")
    }
    launch {
        (1..3).forEach{
            channel.send(it)
            println("发送数据: $it")
        }
        // 关闭channel， 节省资源
        channel.close()
    }
    launch {
        for (i in channel){
            println("接收数据: $i")
        }
    }
    println("结束")
}
输出结果如下：
开始
结束
发送数据: 1
发送数据: 2
发送数据: 3
接收数据: 2
接收数据: 3

val channel: ReceiveChannel<Int> = produce {
        (1..100).forEach{
            send(it)
            println("发送： $it")
        }
    }
while (!channel.isClosedForReceive){
    val i = channel.receive();
    println("接收： $i")
}
输出报错信息：
Exception in thread "main" kotlinx.coroutines.channels.ClosedReceiveChannelException: Channel was closed

val channel: ReceiveChannel<Int> = produce {
        (1..100).forEach{
            send(it)
            println("发送： $it")
        }
    }
channel.consumeEach {
    println("接收：$it")
}

    println("开始")
    val channel = Channel<Int>(capacity = 3, onBufferOverflow = BufferOverflow.DROP_OLDEST) {
        println("onUndeliveredElement = $it")
    }
    launch {
        (1..3).forEach{
            channel.send(it)
            println("发送数据: $it")
        }
    }
    println("结束")
}
输出：
开始
结束
发送数据: 1
发送数据: 2
发送数据: 3