Kotlin之协程（四）协程并发

认识channel

channel是一个并发安全的队列，可以连接协程，实现不同协程的通信。

Library中定义了几种类型的Channel。 它们在内部能够存储多种元素,只是在send调用是否能够挂起方面有所不一样。 对于全部通道类型，receive调用的行为方式相同:若是通道不为空，则接收元素，不然将挂起。

Unlimited channel

无限制通道(Unlimited channel)是最接近队列的模拟:生产者能够将元素发送到此通道，而且它将无限增加。 send方法将永远不会被挂起。 若是没有更多的内存，则会抛出OutOfMemoryException。 和队列不一样的是当使用者尝试从空通道接收消息并被挂起直到有一些新元素发送到该通道时继续使用。

Buffered channel

缓冲通道(Buffered channel)的大小受指定数字的限制。 生产者能够将元素发送到此通道，直到达到最大限制。 全部元素都在内部存储。 通道已满时，下一个send呼叫将被挂起，直到出现更多可用空间。

Rendezvous channel

"约定"通道(Rendezvous channel)是没有缓冲区的通道。 这与建立大小为零的缓冲通道(Buffered channel)相同。 其中一个功能(send或receive)始终被挂起，直到调用另外一个功能为止。 若是调用了send函数，但消费者没有准备好处理该元素则receive会挂起，而且send也会被挂起。 一样，若是调用了receive函数且通道为空,换句话说,没有准备好发送该元素的的send被挂起-receive也会被挂起。

Conflated channel

发送到合并通道( Conflated channel)的新元素将覆盖先前发送的元素，所以接收方将始终仅能获取最新元素。 send调用将永远不会被挂起。
建立通道时，指定其类型或缓冲区大小(若是须要缓冲的通道)：

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

默认状况下，会建立一个"约定"通道(Rendezvous channel)。

在如下示例中，将建立一个"约定"通道，两个生产者协程和一个消费者协程:

import kotlinx.coroutines.channels.Channelimport kotlinx.coroutines.*fun main() = runBlocking { val channel = Channel() launch { channel.send("A1") channel.send("A2") log("A done") } launch { channel.send("B1") log("B done") } launch { repeat(3) { val x = channel.receive() log(x) } }}fun log(message: Any?) { println("[${Thread.currentThread().name}] $message")}以上将会打印以下结果:[main @coroutine#4] A1[main @coroutine#4] B1[main @coroutine#2] A done[main @coroutine#3] B done[main @coroutine#4] A2

channel实际上是一个队列，队列中一定存在缓冲区，这个缓冲区满了并且一直没有人调用receive取走函数，send就需要挂起，故意让接收端的节奏放慢，发现send总是被挂起，直到receive之后才会继续往下执行。

fun run1() { val channel = Channel(Channel.UNLIMITED) //Channel协程间通信，并发安全的队列 runBlocking { //生产者 val p = launch { for (i in 1..5) { channel.send(i) log("send = $i") } } //消费者 val c = launch { //正常接收数据 while (true) { //故意让接收端的节奏放慢，发现send总是被挂起，直到receive之后才会继续往下执行 delay(2000) val el = channel.receive() log("re = $el") } //通过迭代器iterator接收数据 //val iterator = channel.iterator() //while (iterator.hasNext()) { // delay(2000) // log("iterator = ${iterator.next()}") //} } joinAll(p,c) } }打印：com.z.zjetpack V/zx: send = 1com.z.zjetpack V/zx: send = 2com.z.zjetpack V/zx: send = 3com.z.zjetpack V/zx: send = 4com.z.zjetpack V/zx: send = 5com.z.zjetpack V/zx: re = 1com.z.zjetpack V/zx: re = 2com.z.zjetpack V/zx: re = 3com.z.zjetpack V/zx: re = 4com.z.zjetpack V/zx: re = 5

produce 与actor

构造生产者与消费者的便捷方法我们可以通过produce方法启动一个生产者协程，并返回一个reveive channel，其他协程就可以用这个channel来接收数据了。反过来我们可以用actor启动一个消费者协程。

fun run2(){ runBlocking { //快捷创建生产者协程，返回一个接收Channel val receiveChannel = produce { repeat(5){ delay(1000) send(it) } } val job2 = launch { for (i in receiveChannel) { log("receiveChannel = $i") } } job2.join() } runBlocking { //构造消费者的便捷方法 val sendChannel = actor { while (true) { val re = receive() log("re = $re") } } val p = launch { for (i in 1..3) { sendChannel.send(i) } } p.join() } }打印：com.z.zjetpack V/zx: receiveChannel = 0com.z.zjetpack V/zx: receiveChannel = 1com.z.zjetpack V/zx: receiveChannel = 2com.z.zjetpack V/zx: receiveChannel = 3com.z.zjetpack V/zx: receiveChannel = 4com.z.zjetpack V/zx: re = 1com.z.zjetpack V/zx: re = 2com.z.zjetpack V/zx: re = 3

channel的关闭

produce和actor返回的channel都会随着对应的协程执行完毕而关闭，也正式这样，channel才会被称为热数据流.对于一个channel，如果我们调用了它的close方法，它会立即停止接收新元素，它的isClosedForSend会立即返回true，由于channel缓冲区的存在，可能还有一些元素没有被处理完，所以要等所有元素都被读取之后isClosedForReceive才会返回truechannel的生命周期最好由主导方来维护，建议由主导的一方实现关闭。

fun run3(){ runBlocking { val channel = Channel(3) //生产者 launch { List(3){ channel.send(it) log("send = $it") } channel.close() log("isClosedForSend = ${channel.isClosedForSend}") log("isClosedForReceive = ${channel.isClosedForReceive}") } //消费者 launch { for (c in channel) { log("re = $c") delay(1000) } log("消费isClosedForSend = ${channel.isClosedForSend}") log("消费isClosedForReceive = ${channel.isClosedForReceive}") } } }打印：com.z.zjetpack V/zx: send = 0com.z.zjetpack V/zx: send = 1com.z.zjetpack V/zx: send = 2com.z.zjetpack V/zx: isClosedForSend = truecom.z.zjetpack V/zx: isClosedForReceive = falsecom.z.zjetpack V/zx: re = 0com.z.zjetpack V/zx: re = 1com.z.zjetpack V/zx: re = 2com.z.zjetpack V/zx: 消费isClosedForSend = truecom.z.zjetpack V/zx: 消费isClosedForReceive = true

BroadcastChannel

发送端和接收端在channel中存在一对多的场景，虽然有多个接收端，但是同一个元素只会被一个接收端读取到，广播则不同，多个接收端不存在互斥行为。

fun run4() { runBlocking { //直接创建 // val broadcastChannel = BroadcastChannel(Channel.BUFFERED) //broadcast方法创建 val channel = Channel() val broadcastChannel = channel.broadcast(Channel.BUFFERED) //创建3个协程来接收 List(3) { launch { val receiveChannel = broadcastChannel.openSubscription() for (r in receiveChannel) { log("协程 $it, re = $r") } } } launch { List(3) { broadcastChannel.send(it) } broadcastChannel.close() } } }打印：com.z.zjetpack V/zx: 协程 0, re = 0com.z.zjetpack V/zx: 协程 0, re = 1com.z.zjetpack V/zx: 协程 0, re = 2com.z.zjetpack V/zx: 协程 1, re = 0com.z.zjetpack V/zx: 协程 1, re = 1com.z.zjetpack V/zx: 协程 1, re = 2com.z.zjetpack V/zx: 协程 2, re = 0com.z.zjetpack V/zx: 协程 2, re = 1com.z.zjetpack V/zx: 协程 2, re = 2

多路复用 复用多个await

两个api分别从网络和本地获取数据，期望哪个先返回就先用哪个做显示

fun CoroutineScope.getFromLocal() = async { delay(1000) "返回本地数据" } fun CoroutineScope.getFromNet() = async { "返回网络数据" } fun run5() { runBlocking { launch { val local = getFromLocal() val net = getFromNet() val res = select { local.onAwait { it } net.onAwait { it } } log("值 = $res") }.join() } }打印：com.z.zjetpack V/zx: 值 = 返回网络数据

复用多个channel

跟await类似，会接收到最快的那个channel消息

fun run6() { runBlocking { val channels = listOf(Channel(), Channel()) launch { delay(100) channels[0].send(1) } launch { delay(500) channels[1].send(5) } val result = select { channels.forEach { re -> re.onReceive{it} } } log("result = $result") } }打印：com.z.zjetpack V/zx: result = 1

SelectClause

哪些事件可以被select？SelectClause类型
包括：
SelectClause0：对应事件没有返回值，例如 join 没有返回值，对应的 onJoin 就是这个类型，使用时 onJoin 的参数是一个无参函数：

public val onJoin: SelectClause0 runBlocking { val job1 = launch { delay(100) log("job1") } val job2 = launch { delay(10) log("job2") } select { job1.onJoin { log("job1.onJoin") } job2.onJoin { log("job2.onJoin") } } }打印：com.z.zjetpack V/zx: job2com.z.zjetpack V/zx: job2.onJoincom.z.zjetpack V/zx: job1

SelectClause1：对应事件有返回值，前面的 onAwait 和 onReceive 都是此类情况。

public val onAwait: SelectClause1 public val onReceive: SelectClause1

SelectClause2：对应事件有返回值，此外还需要额外的一个参数，例如 Channel.onSend 有两个参数，第一个就是一个 Channel 数据类型的值，表示即将发送的值，第二个是发送成功时的回调。
如果我们想要确认挂起函数是否支持select，查看是否存在对应的SelectClauseN类型可回调即可。

//返回SelectClause2 public val onSend: SelectClause2> runBlocking { val channels = listOf(Channel(), Channel()) launch { select { launch { delay(100) channels[0].onSend(1) { sendChannel -> log("send on $sendChannel") } } launch { delay(500) channels[1].onSend(5) { sendChannel -> log("send on $sendChannel") } } } } launch { for (c in channels) { log("数据 = ${c.receive()}") } } }打印：com.z.zjetpack V/zx: send on RendezvousChannel@63db1bf{EmptyQueue}com.z.zjetpack V/zx: 数据 = 1

Flow实现多路复用

coroutineScope { val login = "..." listOf(::getUserFromApi, ::getUserFromLocal) ..、① .map { function -> function.call(login) ..、② } .map { deferred -> flow { emit(deferred.await()) } ..、③ } .merge() ..、④ .onEach { user -> println("Result: $user") }.launchIn(this)}

这其中，① 处用创建了两个函数引用组成的 List；② 处调用它们得到 deferred；③ 处比较关键，对于每一个 deferred 我们创建一个单独的 Flow，并在 Flow 内部发送 deferred.await() 返回的结果，即返回的 User 对象；现在我们有了两个 Flow 实例，我们需要将它们整合成一个 Flow 进行处理，调用 merge 函数即可。

协程的并发安全

除了线程中常用的解决并发安全问题的手段外，协程提供了一些并发安全的工具

channel：并发安全的消息通道Mutex：轻量级锁，lock和unlock和线程锁类似，轻量级是说它在获取不到锁时不会阻塞线程而是挂起等待锁的释放。Semaphore：轻量级信号量，信号量可以有多个，协程在获取到信号量后即可执行并发操作，当Semaphore的参数为1时，效果等同于Mutex

fun run7() { runBlocking { var count = 0 List(10000) { //GlobalScope是线程不安全的 GlobalScope.launch { count++ } }.joinAll() log("默认count = $count") } //使用volatile解决并发问题 runBlocking { var count = AtomicInteger(0) List(10000) { //GlobalScope是线程不安全的 GlobalScope.launch { count.incrementAndGet() } }.joinAll() log("volatile count = ${count.get()}") } //使用Mutex解决并发问题 runBlocking { var count = 0 var mutex = Mutex() List(10000) { //GlobalScope是线程不安全的 GlobalScope.launch { mutex.withLock { count++ } } }.joinAll() log("Mutex count = $count") } //使用Semaphore解决并发问题 runBlocking { var count = 0 var semaphore = Semaphore(1) List(10000) { //GlobalScope是线程不安全的 GlobalScope.launch { semaphore.withPermit { count++ } } }.joinAll() log("Semaphore count = $count") } }

打印：

com.z.zjetpack V/zx: 默认count = 9991com.z.zjetpack V/zx: volatile count = 10000com.z.zjetpack V/zx: Mutex count = 10000com.z.zjetpack V/zx: Semaphore count = 10000

除了使用这些工具解决并发问题，也可以避免访问外部可变状态，编写函数时，要求它不得访问外部状态，只能基于入参做运算，通过返回值提供运算结果。

runBlocking { var count = 0 //count在协程外面不存在并发问题 val result = count + List(10000){ GlobalScope.async { 1 } }.map { it.await() }.sum() log("count count = $result") }