企业官网建设流程全解析

Channel

类似于BlockingQueue，其中元素被添加到一端并从另一端接收，send和receive方法都声明为suspend

val channel = Channel<Int>() launch { // this might be heavy CPU-consuming computation or async logic, // we'll 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循环遍历channel

val channel = Channel<Int>() launch { for (x in 1..5) channel.send(x * x) channel.close() // we're 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 producers

produce可以构建一个生产者Channel，会自动起协程，并调用close，省去上面的步骤，消费时可通过consumeEach()

fun CoroutineScope.produceSquares(): ReceiveChannel<Int> = 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) = produce<Int> { var x = start while (true) send(x++) // infinite stream of integers from start }

fun CoroutineScope.square(numbers: ReceiveChannel<Int>): ReceiveChannel<Int> = 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 coroutines

Fan-out

一个生产者对应多个消费者

fun CoroutineScope.produceNumbers() = produce<Int> { var x = 1 // start from 1 while (true) { send(x++) // produce next delay(100) // wait 0.1s } }

协程中使用for循环，若一个协程失败，其他协程仍可继续，而consumeEach异常时会关闭channel

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch { for (msg in channel) { println("Processor #$id received $msg") } }

取消生产者协程，会同时关闭channel

val producer = produceNumbers() repeat(5) { launchProcessor(it, producer) } delay(950) producer.cancel() // cancel producer coroutine and thus kill them all

Processor #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 10

Fan-in

多个协程可以发送到同一个通道

suspend fun sendString(channel: SendChannel<String>, s: String, time: Long) { while (true) { delay(time) channel.send(s) } }

val channel = Channel<String>() 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 finish

foo foo BAR! foo foo BAR!

Channel Type

Channel由三个不同的接口表示：SendChannel、ReceiveChannel和Channel，后者扩展了前两个接口。

interface SendChannel<in E> { suspend fun send(element: E) fun close(): Boolean } interface ReceiveChannel<out E> { suspend fun receive(): E } interface Channel<E> : SendChannel<E>, ReceiveChannel<E>

• 无限通道：send调用永远不会被挂起（内存不足将收到OutOfMemoryException），从空的通道receive时会挂起直到有新元素
• 缓冲通道：send调用达到大小限制时会被挂起，直到有空间
• 会合通道：send或receive总是挂起，直到另一个被调用
• 汇流通道：新元素将覆盖之前发送的元素，send调用永远不会挂起，receiver始终获得最新的元素

val rendezvousChannel = Channel<String>() val bufferedChannel = Channel<String>(10) val conflatedChannel = Channel<String>(CONFLATED) val unlimitedChannel = Channel<String>(UNLIMITED)

fun runChannelExample() = runBlocking { println("——————————————————————————————————————————————————") val channel = Channel<String>() // 每道菜的做菜耗时不同，用来演示"谁先做好谁先上" 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

每次经过给定的延迟时都会产生Unit

fun main() = runBlocking<Unit> { 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-200=100ms

Initial 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