kotlin--Channel、多路复用、并发安全

一、Channel

Channel相较于Flow，Flow是冷流，本质上可以说是一个单线程操作，只有开始收集时，上流代码才会启动，而Channel是一个并发安全的队列，可以用来连接不同的协程，实现不同协程之间的通信

1.Channel的使用

创建一个Channel对象，在不同协程中调用其send和receive函数

fun `test channel`() = runBlocking {val channel = Channel<Int>()var isFinished = false//生产者val producer = GlobalScope.launch {var i = 0while (i < 3) {delay(200)//发送产品channel.send(++i)}isFinished = true}//消费者val consumer = GlobalScope.launch {while (!isFinished) {//接收产品println(channel.receive())}}joinAll(producer, consumer)
}

结果：
1
2
3

2.Channel的容量

Channel的容量或者说缓冲区大小，默认为0，当消费者消费慢了，那么生产者会等待，反之生产者生产慢了，消费者会等待。如果想要指定缓冲区大小，可以在构建时传入

3.Channel迭代器

除了使用receive函数外，Channel还提供了迭代器用来接收数据

fun `test channel iterator`() = runBlocking {val channel = Channel<Int>(Channel.UNLIMITED)val producer = GlobalScope.launch {for (i in 0..5) {channel.send(i)}}val consumer = GlobalScope.launch { val iterator = channel.iterator()while(iterator.hasNext()){val element = iterator.next()println(element)}//        for(value in channel){
//            println(value)
//        }}joinAll(producer,consumer)
}

4.produce与actor

在协程中，可以使用produce启动一个生产者协程，并返回ReceiveChannel

fun `test channel produce`() = runBlocking {val receiveChannel = GlobalScope.produce {repeat(3){delay(100)send(it)}}val consumer = GlobalScope.launch { for(i in receiveChannel){println(i)}}consumer.join()
}

反之使用actor启动一个消费者协程

fun `test channel acotr`() = runBlocking {val sendChannel = GlobalScope.actor<Int> {while (true) {delay(100)println(receive())}}val producer = GlobalScope.launch {for (i in 0..3) {sendChannel.send(i)}}producer.join()
}

5.Channel的关闭

produce和actor返回的channel都会随着对应协程执行结束后自动关闭
我们也可以使用close方法手动关闭，它会立即停止发送元素，此时isClosedForSend会立即返回true，而由于缓冲区的存在，所有元素读取完毕后，isClosedForReceive才会返回true

fun `test channel close`() = runBlocking { val channel = Channel<Int>(3)val produce = GlobalScope.launch {for (i in 1..3) {channel.send(i)}channel.close()println("produce isClosedForSend: ${channel.isClosedForSend}" +" isClosedForReceive: ${channel.isClosedForReceive}")}val consumer = GlobalScope.launch {for (i in channel) {delay(100)println(i)}println("consumer isClosedForSend: ${channel.isClosedForSend}" +"isClosedForReceive: ${channel.isClosedForReceive}")}joinAll(produce, consumer)
}

结果：
produce isClosedForSend: true isClosedForReceive: false
1
2
3
consumer isClosedForSend: trueisClosedForReceive: true

6.BroadcastChannel

kotlin还提供了发送端一对多接收端的方式，使用BroadcastChannel需要指定其缓冲区大小，或使用Channel.BUFFERED。还可以使用channel对象的broadcast函数来获取BroadcastChannel对象

fun `test channel broadcast`() = runBlocking {
//    val broadcastChannel = BroadcastChannel<Int>(Channel.BUFFERED)val channel = Channel<Int>(3)val broadcastChannel = channel.broadcast()GlobalScope.launch {(0..3).forEach {broadcastChannel.send(it)}broadcastChannel.close()}List(3) { index ->GlobalScope.launch {val channel = broadcastChannel.openSubscription()for (i in channel) {println("index : $index  receive: $i")}}}.joinAll()
}

结果：
index : 1 receive: 0
index : 1 receive: 1
index : 1 receive: 2
index : 1 receive: 3
index : 0 receive: 0
index : 2 receive: 0
index : 2 receive: 1
index : 2 receive: 2
index : 2 receive: 3
index : 0 receive: 1
index : 0 receive: 2
index : 0 receive: 3

Process finished with exit code 0

二、多路复用

数据通信系统或计算机网络系统中，传输媒体的带宽或容量往往会大于传输单一信号的需求，为了有效利用通道线路，希望一个信道同时传输多路信号，这就是多路复用技术

1.多个await复用--select

获取数据可能有多个方法，每个方法的耗时可能不太一样，调用这些方法后，我们希望哪个方法先返回就使用它的返回值

data class Response<T>(var data: T, var random: Long)fun CoroutineScope.getInfoForLocal1() = async {// 随机一个等待值val random = (0..201).shuffled().first().toLong()delay(random)Response("info for local1", random)
}fun CoroutineScope.getInfoForLocal2() = async {// 随机一个等待值val random = (0..201).shuffled().first().toLong()delay(random)Response("info for local2", random)
}fun `test await select`() = runBlocking {val local1 = getInfoForLocal1()val local2 = getInfoForLocal2()val select = select<Response<String>> {local1.onAwait { response ->response //可以改成其他类型的}local2.onAwait { response ->response}}println("$select")
}

结果：
Response(data=info for local1, random=56)

2.多个Channel复用

和await差不多

fun `test channel select`() = runBlocking {val channels = listOf(Channel<Int>(), Channel<Int>())GlobalScope.launch {delay(50)channels[0].send(0)}GlobalScope.launch {delay(100)channels[1].send(1)}val select = select<Int> {channels.forEach {it.onReceive { value ->value}}}println(select)delay(1000)
}

结果
0

3.SelectCause

并不是所有的事件可以使用select的，只有SelectCauseN类型的事件
1.SelectCause0：对应事件没有返回值，例如join，那么onJoin就是SelectCauseN，使用时，onJoin的参数是一个无参函数
2.SelectCause1：对应事件有返回值，例如onAwait，onReceive
3.SelectCause3：对应事件有返回值，此外还要一个额外参数，例如Channel.onSend，一个参数为Channel数据类型的值，一个为发送成功时的回调

SelectCause3:

fun `test channel send select`() = runBlocking {val channels = listOf(Channel<Int>(), Channel<Int>())launch {select<Unit?> {launch {delay(100)channels[0].onSend(0) {println("onSend 0")}}launch {delay(50)channels[1].onSend(1) {println("onSend 1")}}}}GlobalScope.launch {println(channels[0].receive())}GlobalScope.launch {println(channels[1].receive())}}

结果：
1
onSend 1

4.使用Flow实现多路复用

fun `test flow merge`() = runBlocking {listOf(::getInfoForLocal1, ::getInfoForLocal2).map { function ->function.call()}.map { deferred ->flow { emit(deferred.await()) }}.merge().collect {println(it)}
}

结果：
Response(data=info for local1, random=129)
Response(data=info for local2, random=145)

和select不同的是，Flow收集时，会收集所有结果

三、并发安全

在Java平台上的kotlin协程实现避免不了并发调度的问题，因此线程安全值得留意

fun `test sync safe1`() = runBlocking {var count = 0;List(1000) {GlobalScope.launch { count++ }}.joinAll()println(count)
}

结果：
987

当然我们还可以使用Java是提供的线程安全类

fun `test sync safe2`() = runBlocking {var count = AtomicInteger(0);List(1000) {GlobalScope.launch { count.incrementAndGet() }}println(count.get())
}

协程框架也提供了解决并发问题的方法：

1.上面学习的Channel

2.Mutex：轻量级锁，用法和Java的锁类似，获取不到锁时，不会阻塞线程，而是挂起等待锁的释放

fun `test sync mutex`() = runBlocking {var count = 0;val mutex = Mutex()List(1000) {GlobalScope.launch {mutex.withLock {count++}}}.joinAll()println(count)
}

3.Semaphore，轻量级信号量，在linux中是进程间通讯的一种方式，协程获取到信号量后即可执行并发操作。当Semaphore为1时，效果等价于Mutex

fun `test sync semaphore`() = runBlocking {var count = 0;val semaphore = Semaphore(1)List(1000) {GlobalScope.launch {semaphore.withPermit {count++}}}.joinAll()println(count)
}

4.我们也可以避免访问外部变量，基于参数作运算，通过返回值提供运算结果

fun `test sync avoid`() = runBlocking {var count = 0;count += List(1000) {GlobalScope.async {1}}.map {it.await()}.sum()println(count)
}