Channel类似于BlockingQueue其中元素被添加到一端并从另一端接收send和receive方法都声明为suspendval channel ChannelInt() launch { // this might be heavy CPU-consuming computation or async logic, // well just send five squares for (x in 1..5) channel.send(x * x) } // here we print five received integers: repeat(5) { println(channel.receive()) } println(Done!)1 4 9 16 25 Done!Closing and iteration over channels使用close后可以用for循环遍历channelval channel ChannelInt() launch { for (x in 1..5) channel.send(x * x) channel.close() // were done sending } // here we print received values using for loop (until the channel is closed) for (y in channel) println(y) println(Done!)1 4 9 16 25 Done!Building channel producersproduce可以构建一个生产者Channel会自动起协程并调用close省去上面的步骤消费时可通过consumeEach()fun CoroutineScope.produceSquares(): ReceiveChannelInt produce { for (x in 1..5) send(x * x) } fun main() runBlocking { val squares produceSquares() squares.consumeEach { println(it) } println(Done!)Pipeline管道用协程产生无限的值流fun CoroutineScope.numbersFrom(start: Int) produceInt { var x start while (true) send(x) // infinite stream of integers from start }fun CoroutineScope.square(numbers: ReceiveChannelInt): ReceiveChannelInt produce { for (x in numbers) send(x * x) }如上使用埃氏筛法产生一条无限数字流每发现一个质数就在链末端接一节「过滤这个质数倍数」的协程层层筛下去val numbers produceNumbers() // produces integers from 1 and on val squares square(numbers) // squares integers repeat(5) { println(squares.receive()) // print first five } println(Done!) // we are done coroutineContext.cancelChildren() // cancel children coroutinesFan-out一个生产者对应多个消费者fun CoroutineScope.produceNumbers() produceInt { var x 1 // start from 1 while (true) { send(x) // produce next delay(100) // wait 0.1s } }协程中使用for循环若一个协程失败其他协程仍可继续而consumeEach异常时会关闭channelfun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannelInt) launch { for (msg in channel) { println(Processor #$id received $msg) } }取消生产者协程会同时关闭channelval producer produceNumbers() repeat(5) { launchProcessor(it, producer) } delay(950) producer.cancel() // cancel producer coroutine and thus kill them allProcessor #2 received 1 Processor #4 received 2 Processor #0 received 3 Processor #1 received 4 Processor #3 received 5 Processor #2 received 6 Processor #4 received 7 Processor #0 received 8 Processor #1 received 9 Processor #3 received 10Fan-in多个协程可以发送到同一个通道suspend fun sendString(channel: SendChannelString, s: String, time: Long) { while (true) { delay(time) channel.send(s) } }val channel ChannelString() launch { sendString(channel, foo, 200L) } launch { sendString(channel, BAR!, 500L) } repeat(6) { // receive first six println(channel.receive()) } coroutineContext.cancelChildren() // cancel all children to let main finishfoo foo BAR! foo foo BAR!Channel TypeChannel由三个不同的接口表示SendChannel、ReceiveChannel和Channel后者扩展了前两个接口。interface SendChannelin E { suspend fun send(element: E) fun close(): Boolean } interface ReceiveChannelout E { suspend fun receive(): E } interface ChannelE : SendChannelE, ReceiveChannelE无限通道send调用永远不会被挂起内存不足将收到OutOfMemoryException从空的通道receive时会挂起直到有新元素缓冲通道send调用达到大小限制时会被挂起直到有空间会合通道send或receive总是挂起直到另一个被调用汇流通道新元素将覆盖之前发送的元素send调用永远不会挂起receiver始终获得最新的元素val rendezvousChannel ChannelString() val bufferedChannel ChannelString(10) val conflatedChannel ChannelString(CONFLATED) val unlimitedChannel ChannelString(UNLIMITED)fun runChannelExample() runBlocking { println(——————————————————————————————————————————————————) val channel ChannelString() // 每道菜的做菜耗时不同用来演示谁先做好谁先上 val dishes listOf(番茄炒蛋 to 300L, 红烧肉 to 100L, 青菜 to 200L) // ① 三个厨师并发做菜做好就放上传送带 for ((name, time) in dishes) { launch { delay(time) // 模拟做菜耗时并发不互相等 channel.send(name) println(厨房$name 做好了) } } // ② 服务员逐个收传送带空就挂起等待有菜就拿 repeat(dishes.size) { val dish channel.receive() println(上菜$dish) } println(全部上齐) println(——————————————————————————————————————————————————) }厨房红烧肉 做好了 上菜红烧肉 厨房青菜 做好了 上菜青菜 厨房番茄炒蛋 做好了 上菜番茄炒蛋 全部上齐Ticker channels每次经过给定的延迟时都会产生Unitfun main() runBlockingUnit { val tickerChannel ticker(delayMillis 200, initialDelayMillis 0) // create a ticker channel var nextElement withTimeoutOrNull(1) { tickerChannel.receive() } println(Initial element is available immediately: $nextElement) // no initial delay nextElement withTimeoutOrNull(100) { tickerChannel.receive() } // all subsequent elements have 200ms delay println(Next element is not ready in 100 ms: $nextElement) nextElement withTimeoutOrNull(120) { tickerChannel.receive() } println(Next element is ready in 200 ms: $nextElement) // Emulate large consumption delays println(Consumer pauses for 300ms) delay(300) // Next element is available immediately nextElement withTimeoutOrNull(1) { tickerChannel.receive() } println(Next element is available immediately after large consumer delay: $nextElement) // Note that the pause between receive calls is taken into account and next element arrives faster nextElement withTimeoutOrNull(120) { tickerChannel.receive() } println(Next element is ready in 100ms after consumer pause in 300ms: $nextElement) tickerChannel.cancel() // indicate that no more elements are needed }Ticker会感知暂停调整下一个元素的延迟停了 300ms超过了一个 200ms 周期下次receive会立刻发送且下下次等待时间变成300-200100msInitial element is available immediately: kotlin.Unit Next element is not ready in 100 ms: null Next element is ready in 200 ms: kotlin.Unit Consumer pauses for 300ms Next element is available immediately after large consumer delay: kotlin.Unit Next element is ready in 100ms after consumer pause in 300ms: kotlin.Unit 
长河君 · 2026/11/4 7:10:46 · 阅读 1.2k Kotlin中级——Channel
Channel类似于BlockingQueue其中元素被添加到一端并从另一端接收send和receive方法都声明为suspendval channel ChannelInt() launch { // this might be heavy CPU-consuming computation or async logic, // well just send five squares for (x in 1..5) channel.send(x * x) } // here we print five received integers: repeat(5) { println(channel.receive()) } println(Done!)1 4 9 16 25 Done!Closing and iteration over channels使用close后可以用for循环遍历channelval channel ChannelInt() launch { for (x in 1..5) channel.send(x * x) channel.close() // were done sending } // here we print received values using for loop (until the channel is closed) for (y in channel) println(y) println(Done!)1 4 9 16 25 Done!Building channel producersproduce可以构建一个生产者Channel会自动起协程并调用close省去上面的步骤消费时可通过consumeEach()fun CoroutineScope.produceSquares(): ReceiveChannelInt produce { for (x in 1..5) send(x * x) } fun main() runBlocking { val squares produceSquares() squares.consumeEach { println(it) } println(Done!)Pipeline管道用协程产生无限的值流fun CoroutineScope.numbersFrom(start: Int) produceInt { var x start while (true) send(x) // infinite stream of integers from start }fun CoroutineScope.square(numbers: ReceiveChannelInt): ReceiveChannelInt produce { for (x in numbers) send(x * x) }如上使用埃氏筛法产生一条无限数字流每发现一个质数就在链末端接一节「过滤这个质数倍数」的协程层层筛下去val numbers produceNumbers() // produces integers from 1 and on val squares square(numbers) // squares integers repeat(5) { println(squares.receive()) // print first five } println(Done!) // we are done coroutineContext.cancelChildren() // cancel children coroutinesFan-out一个生产者对应多个消费者fun CoroutineScope.produceNumbers() produceInt { var x 1 // start from 1 while (true) { send(x) // produce next delay(100) // wait 0.1s } }协程中使用for循环若一个协程失败其他协程仍可继续而consumeEach异常时会关闭channelfun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannelInt) launch { for (msg in channel) { println(Processor #$id received $msg) } }取消生产者协程会同时关闭channelval producer produceNumbers() repeat(5) { launchProcessor(it, producer) } delay(950) producer.cancel() // cancel producer coroutine and thus kill them allProcessor #2 received 1 Processor #4 received 2 Processor #0 received 3 Processor #1 received 4 Processor #3 received 5 Processor #2 received 6 Processor #4 received 7 Processor #0 received 8 Processor #1 received 9 Processor #3 received 10Fan-in多个协程可以发送到同一个通道suspend fun sendString(channel: SendChannelString, s: String, time: Long) { while (true) { delay(time) channel.send(s) } }val channel ChannelString() launch { sendString(channel, foo, 200L) } launch { sendString(channel, BAR!, 500L) } repeat(6) { // receive first six println(channel.receive()) } coroutineContext.cancelChildren() // cancel all children to let main finishfoo foo BAR! foo foo BAR!Channel TypeChannel由三个不同的接口表示SendChannel、ReceiveChannel和Channel后者扩展了前两个接口。interface SendChannelin E { suspend fun send(element: E) fun close(): Boolean } interface ReceiveChannelout E { suspend fun receive(): E } interface ChannelE : SendChannelE, ReceiveChannelE无限通道send调用永远不会被挂起内存不足将收到OutOfMemoryException从空的通道receive时会挂起直到有新元素缓冲通道send调用达到大小限制时会被挂起直到有空间会合通道send或receive总是挂起直到另一个被调用汇流通道新元素将覆盖之前发送的元素send调用永远不会挂起receiver始终获得最新的元素val rendezvousChannel ChannelString() val bufferedChannel ChannelString(10) val conflatedChannel ChannelString(CONFLATED) val unlimitedChannel ChannelString(UNLIMITED)fun runChannelExample() runBlocking { println(——————————————————————————————————————————————————) val channel ChannelString() // 每道菜的做菜耗时不同用来演示谁先做好谁先上 val dishes listOf(番茄炒蛋 to 300L, 红烧肉 to 100L, 青菜 to 200L) // ① 三个厨师并发做菜做好就放上传送带 for ((name, time) in dishes) { launch { delay(time) // 模拟做菜耗时并发不互相等 channel.send(name) println(厨房$name 做好了) } } // ② 服务员逐个收传送带空就挂起等待有菜就拿 repeat(dishes.size) { val dish channel.receive() println(上菜$dish) } println(全部上齐) println(——————————————————————————————————————————————————) }厨房红烧肉 做好了 上菜红烧肉 厨房青菜 做好了 上菜青菜 厨房番茄炒蛋 做好了 上菜番茄炒蛋 全部上齐Ticker channels每次经过给定的延迟时都会产生Unitfun main() runBlockingUnit { val tickerChannel ticker(delayMillis 200, initialDelayMillis 0) // create a ticker channel var nextElement withTimeoutOrNull(1) { tickerChannel.receive() } println(Initial element is available immediately: $nextElement) // no initial delay nextElement withTimeoutOrNull(100) { tickerChannel.receive() } // all subsequent elements have 200ms delay println(Next element is not ready in 100 ms: $nextElement) nextElement withTimeoutOrNull(120) { tickerChannel.receive() } println(Next element is ready in 200 ms: $nextElement) // Emulate large consumption delays println(Consumer pauses for 300ms) delay(300) // Next element is available immediately nextElement withTimeoutOrNull(1) { tickerChannel.receive() } println(Next element is available immediately after large consumer delay: $nextElement) // Note that the pause between receive calls is taken into account and next element arrives faster nextElement withTimeoutOrNull(120) { tickerChannel.receive() } println(Next element is ready in 100ms after consumer pause in 300ms: $nextElement) tickerChannel.cancel() // indicate that no more elements are needed }Ticker会感知暂停调整下一个元素的延迟停了 300ms超过了一个 200ms 周期下次receive会立刻发送且下下次等待时间变成300-200100msInitial element is available immediately: kotlin.Unit Next element is not ready in 100 ms: null Next element is ready in 200 ms: kotlin.Unit Consumer pauses for 300ms Next element is available immediately after large consumer delay: kotlin.Unit Next element is ready in 100ms after consumer pause in 300ms: kotlin.Unit