【深入理解Kotlin协程】协程中的Channel和Flow 协程中的线程安全问题

热数据通道 Channel

在这里插入图片描述

Channel 实际上就是个并发安全的队列，它可以用来连接协程，实现不同协程的通信，代码如代码清单所示

suspend fun testChannel() {
    val channel = Channel<Int>()  
    var i = 0
    //生产者 发
    val producer = GlobalScope.launch {
        while (true) {
            delay(1000)
            channel.send(i++)
        }
    }
    //消费者 收
    val consumer = GlobalScope.launch {
        while (true) {
            val value = channel.receive()
            println("received <<<<<<<<<<<<<<<<<< $value")
        }
    }
    producer.join()
    consumer.join()
}

上述代码构造了两个协程 producer 和 consumer, 没有为它们明确指定调度器，所以它们都是采用默认调度器，在 Java 平台上就是基于线程池实现的 Default。它们可以运行在不同的线程上，也可以运行在同一个线程上，具体执行流程如图 6-2 所示。

producer 每隔 1s 向 Channel 发送 1 个数，consumer 一直读取 channel 来获取这个数字并打印，显然发送端比接收端更慢，在没有值可以读到的时候， receive 是挂起的，直到有新元素到达。

这么看来，receive 一定是一个挂起函数，那么 send 呢？

你会发现 send 也是挂起函数。发送端为什么会挂起？以我们熟知的 BlockingQueue 为例，当我们往其中添加元素的时候，元素在队列里实际上是占用了空间的，如果这个队列空间不足，那么再往其中添加元素的时候就会出现两种情况：

阻塞．等待队列腾出空间
异常，拒绝添加元素。

send 也会面临同样的问题， Channel 实际上就是个队列，队列中一定存在缓冲区，那么一旦这个缓冲区满了，并且也一直没有人调用 receive 并取走元素， send 就需要挂起，等待接收者取走数据之后再写入 Channel 。

Channel缓冲区

public fun <E> Channel(
    capacity: Int = RENDEZVOUS,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    onUndeliveredElement: ((E) -> Unit)? = null
): Channel<E> =
    when (capacity) {
        RENDEZVOUS -> {
            if (onBufferOverflow == BufferOverflow.SUSPEND)
                RendezvousChannel(onUndeliveredElement) // an efficient implementation of rendezvous channel
            else
                ArrayChannel(1, onBufferOverflow, onUndeliveredElement) // support buffer overflow with buffered channel
        }
        CONFLATED -> {
            require(onBufferOverflow == BufferOverflow.SUSPEND) {
                "CONFLATED capacity cannot be used with non-default onBufferOverflow"
            }
            ConflatedChannel(onUndeliveredElement)
        }
        UNLIMITED -> LinkedListChannel(onUndeliveredElement) // ignores onBufferOverflow: it has buffer, but it never overflows
        BUFFERED -> ArrayChannel( // uses default capacity with SUSPEND
            if (onBufferOverflow == BufferOverflow.SUSPEND) CHANNEL_DEFAULT_CAPACITY else 1,
            onBufferOverflow, onUndeliveredElement
        )
        else -> {
            if (capacity == 1 && onBufferOverflow == BufferOverflow.DROP_OLDEST)
                ConflatedChannel(onUndeliveredElement) // conflated implementation is more efficient but appears to work in the same way
            else
                ArrayChannel(capacity, onBufferOverflow, onUndeliveredElement)
        }
    }

我们构造 Channel 的时候调用了一个名为 Channel 的函数，虽然两个 “Channel" 起来是样的，但它却确实不是 Channel 的构造函数。在 Kotlin 中我们经常定义一个顶级函数来伪装成同名类型的构造器，这本质上就是工厂函数。Channel 函数有一个参数叫 capacity, 该参数用于指定缓冲区的容量，RENDEZVOUS 默认值为 0，RENDEZVOUS 本意就是描述“不见不散＂的场景，如果不调用 receive, send 就会一直挂起等待。如果把上面代码中consumer的channel.receive()注释掉，则producer中send方法第一次调用就会挂起。

Channel(Channel.RENDEZVOUS ) 的方式是有人接收才会继续发，边收边发，如果没有接受的，则发送者会挂起等待。
Channel(Channel.UNLIMITED ) 的方式是发送者发送完毕，就直接返回，不管有没有接受者。
Channel(Channel.CONFLATED ) 的方式是不管发送者发了多少个，接受者只能收到最后一个，也是发送完就返回了，不管有没有接受者。
Channel(Channel.BUFFERED ) 的方式也是发送者发送完就返回了，不管有没有接受者，可以指定buffer大小。
Channel(1) 的方式指定管道的容量大小，如果数据超过容量，发送者就会挂起等待，直到有接受者取走数据，发送者才发送下一批数据。

Channel的迭代

Channel可以通过迭代器迭代访问：

GlobalScope.launch {
    val iterator = channel.iterator()
    while (iterator.hasNext()) { // 挂起点
          println("received <<<<<<<<<<<<<<<<<< ${iterator.next()}")
    }
}

其中，iterator.hasNext() 是挂起函数，在判断是否有下个元素的时候就需要去Channel 中读取元素了，这个写法自然可以简化成 for-in。

GlobalScope.launch {
    for (element in channel) { 
          println("received <<<<<<<<<<<<<<<<<< $element")
    }
}

生产者和消费者协程构造器

我们可以通过 produce 方法启动一个生产者协程，并返回 ReceiveChannel，其他协程就可以用这个 Channel 来接收数据了。反过来，我们可以用 actor 启动一个消费者协程。

suspend fun producer() {
    val receiveChannel = GlobalScope.produce {
        for (i in 0..3) {
            send(i)
            println("send --------------> $i")
        }
    }

    val consumer = GlobalScope.launch {
        for (i in receiveChannel) {
            println("received <<<<<<<<<<<<<<<< $i")
        }
    }

    consumer.join()
}

suspend fun consumer() {
    val sendChannel = GlobalScope.actor<Int> {
        for (i in this) {
            println("received <<<<<<<<<<<<<<<< $i")
        }
    }

    val producer = GlobalScope.launch {
        for (i in 0..3) {
            sendChannel.send(i)
            println("send --------------> $i")
        }
    }

    producer.join()
}

使用这两种构造器也可以指定Channel对应的缓冲区类型，如：

val receiveChannel = GlobalScope.produce(capacity = Channel.UNLIMITED) {
    for (i in 0..3) {
        send(i)
    }
}

ReceiveChannel SendChannel 都是 Channel 的父接口，前者定义了 receive, 后者定义了 send, Channel 也因此既可以使用 receive 又可以使用 send。
通过 produce 和 actor 这两个协程构造器启动的协程也与返回的 Channel 自然地绑定到了一起，因此在协程结束时返回的 Channel 也会被立即关闭。

以 produce 为例，它构造出了一个 ProducerCoroutine 对象，该对象也是 Job 的实现：

private class ProducerCoroutine<E>(
    parentContext: CoroutineContext, channel: Channel<E>
) : ChannelCoroutine<E>(parentContext, channel, true, active = true), ProducerScope<E> {
    override val isActive: Boolean
        get() = super.isActive

    override fun onCompleted(value: Unit) {
        _channel.close() // 协程完成时关闭channel
    }

    override fun onCancelled(cause: Throwable, handled: Boolean) {
        val processed = _channel.close(cause) // 协程取消时关闭channel
        if (!processed && !handled) handleCoroutineException(context, cause)
    }
}

注意，在协程完成和取消的方法调用中，对应的＿channel 都会被关闭。produc actor 这两个构造器看上去都很有用，不过目前前者仍被标记为 Experimental CoroutinesApi, 后者则被标记为 ObsoleteCoroutinesApi, 后续仍然可能会有较大的改动。

Channel的关闭

对千一个 Channel 如果我们调用了它的 close() 方法，它会立即停止接收新元素，也就是说这时候它的 isClosedForSend 会立即返回 true 而由于 Channel 缓冲区的存在，这时候可能还有一些元素没有被处理完，因此要等所有的元素都被读取之后 isClosedForReceive 才会返回 true。

一说到关闭，我们很容易想到 I/0, 如果不关闭 1/0 可能会造成资源泄露。那么 Channel 关闭有什么意义呢？前面我们提到过，Channel 内部的资源其实就是个缓冲区，如果我们创建 Channel 而不去关闭它。虽然并不会造成系统资源的泄露，但却会让接收端一直处千挂起等待的状态，因此一定要在适当的时机关闭 Channel。

究竟由谁来关闭Channel，需要根据业务场景由发送端和接受端之间进行协商决定。如果发送端关闭了Channel，接受端还在调用receive方法，会导致异常，这时就需要进行异常处理：

suspend fun testChannel2() {
    val channel = Channel<Int>()
    //生产者 发
    val producer = GlobalScope.launch {
        for (i in 0..3) {
            println("sending --------------> $i")
            channel.send(i)
        }
        channel.close() // 发送端关闭channel
    }
    //消费者 收
    val consumer = GlobalScope.launch {
        try {
            while (true) {
                val value = channel.receive()
//                 val value = channel.receiveCatching() // 这个方法不会抛出异常
                println("received <<<<<<<<<<<<<<<<<< $value")
            }
        } catch (e : ClosedReceiveChannelException) {
            println("catch ClosedReceiveChannelException: ${e.message}")
        }
    }
    producer.join()
    consumer.join()
}

发送端关闭了Channel，接受端还在调用receive方法，会抛出ClosedReceiveChannelException异常，如果使用receiveCatching()遇到close时就不会抛出异常，但是会使用null作为返回结果。

BroadcastChannel

创建 broadcastCbannel 的方法与创建普通的 Channel 几乎没有区别：

val broadcastChannel = broadcastChannel<Int>(5)

如果要订阅功能，那么只要调用如下方法

val receiveChannel = broadcastChannel.openSubscription()

这样我们就得到了一个 ReceiveChannel，如果想要想获取订阅的消息，只需要调用它的 receive 函数；如果想要取消订阅则调用 cancel 函数即可。

我们来看一个比较完整的例子，本示例中我们在发送端发 0 1 2, 并启动 3个协程同时接收广播，相关代码如下所示。

suspend fun broadcast() {
    //下面几种都可以创建一个BroadcastChannel
    //val broadcastChannel = BroadcastChannel<Int>(Channel.BUFFERED)
    //val broadcastChannel = Channel<Int>(Channel.BUFFERED).broadcast()
    val broadcastChannel = GlobalScope.broadcast {
        for (i in 0..2) {
            send(i)
        }
    }
    //启动3个子协程作为接受者，每个都能收到
    List(3) { index ->
        GlobalScope.launch {
            val receiveChannel = broadcastChannel.openSubscription() // 订阅
            for (i in receiveChannel) {
                println("[#$index] received: $i")
            }
        }
    }.joinAll()
}

除了直接创建以外，我们也可以用前面定义的普通 Channel 进行转换，代码如下所示。

// 通过 Channel 实例直接创建广播
val channel = Channel<Int>()
val broadcastChannel = channel.broadcast()

Channel 版本的序列生成器

// 使用channel模拟序列生成器
val channel = GlobalScope.produce {
    println("A")
    send(1)
    println("B")
    send(2)
    println("Done")
}
for (item in channel) {
    println("get $item")
}

冷数据流Flow

Sequence中不能调用其他挂起函数，不能设置调度器，只能单线程中使用。而Flow可以支持：

// 序列生成器中不能调用其他挂起函数
sequence {
    (1..3).forEach {
        yield(it)
        delay(100) // ERROR
    }
}

创建Flow

val intFlow = flow {
   (1..3).forEach {
       emit(it)
       delay(100)
   }
}

Flow 也可以设定它运行时所使用的调度器：

intFlow.flowDn(Dispatchers.IO)

通过 flowOn 设置的调度器只对它之前的操作有影响，因此这里意味着 intFlow 的构造逻辑会在 IO 调度器上执行。
在这里插入图片描述

最终读取 intFlow 需要调用 collect 函数，这个函数也是一个挂起函数。我们启动一个协程来消费 intFlow, 代码如下所示

suspend fun testFlows(){
    val dispatcher = Executors.newSingleThreadExecutor {
        Thread(it, "MyThread").also { it.isDaemon = true }
    }.asCoroutineDispatcher()

    GlobalScope.launch(dispatcher) {
        val intFlow = flow {
            (1..3).forEach {
                emit(it)
                println("${Thread.currentThread().name}: emit $it")
                delay(1000)
            }
        }
        intFlow.flowOn(Dispatchers.IO)
            .collect {
                println("${Thread.currentThread().name}: collect $it")
            }
    }.join()
}

为了方便区分，我们为协程设置了一个自定义的调度器，它会将协程调度到名叫 MyThread 的线程上，结果如下：

也就是说，collect中的代码运行在Global.launch指定的调度器上，flow{…} 中的代码运行在 flowOn 指定的调度器上。

对比 RxJava 的线程切换

RxJava 也是个基千响应式编程模型的异步框架，它提供了两个切换调度器的 API, 分别是 subscribeOn observeOn, 其中 subscribeOn 指定的调度器执行被观察者的代码， observeOn 指定调度器运行观察者的代码，观察者最后在observeOn 指定调度器上收集结果。flowOn 就相当于subscribeOn，而 launch的调度器 就相当于 observeOn 指定的调度器。

在一个 Flow 创建出来之后，不消费则不生产，多次消费则多次生产，生产和消费总是相对应的，代码如下所示。

GlobalScope.launch(dispatcher) {
    val intFlow = flow {
        (1..3).forEach {
            emit(it)
            delay(1000)
        }
    }
    // Flow 可以被重复消费
    intFlow.collect { println(it) }
    intFlow.collect { println(it) }
}.join()

消费它会输出 “1, 2, 3", 重复消费它会重复输出“1, 2, 3"。

这一点类似于我们前面提到的序列生成器和 RxJava 的例子，它们也都有自己的消费端。我们创建序列后去迭代它，每次迭代都会创建一个新的迭代器从头开始迭代RxJava Observable 也是如此，每次调用它的 subscribe 都会重新消费一次。

所谓冷数据流，就是只有消费时才会生的数据流，这一点 Channel 正好相反，Channel 发送端并不依赖于接收端。

异常处理

Flow 的异常处理也比较接，直接调用 catch 函数即可，如下所示。

如果想要在 Flow 完成时执行逻辑，可以使用 onCompletion，onCompletion 用起来类似于 try… catch… finally 中的 finally, 无论前面是否存在异常，它都会被调用，参数 t 则是前面未捕获的异常。

这套处理机制的设计初衷是确保 Flow 操作中异常的透明。因此，直接使用 try - catch - finally 的写法是违反 Flow 的设计原则的。

suspend fun exception(){
    flow<Int> {
        emit(1)
        throw ArithmeticException("Div 0")
    }.catch {t: Throwable ->
        log("caught error: $t")
    }.onCompletion { t: Throwable? ->
        log("finally.")
    }.flowOn(Dispatchers.IO)
        .collect { log(it) }
// 不推荐直接使用 try - catch - finally 写法
//    flow { // bad!!!
//        try {
//            emit(1)
//            throw ArithmeticException("Div 0")
//        } catch (t: Throwable){
//            log("caught error: $t")
//        } finally {
//            log("finally.")
//        }
//    }
}

我们在 Flow 操作内部使用 try… catch. … finally, 这样的写法后续可能会被禁用。

// Flow 从异常中恢复
flow {
    emit(1)
    throw ArithmeticException("divide 0")
}.catch { t : Throwable ->
    println("caught error $t")
    emit(10)
}

这里我们可以使用 emit 重新生产新元素。细心的读者一定会发现， emit 定义在 FlowCollector 中，因此只要遇到 Receiver 为 FlowCollector 的函数，我们就可以生产新元素。

末端操作符

前面的例子中，我们用 collect 消费 flow 的数据 collect 是最基本的末端操作符，功能与 RxJava 的 subscribe 类似。

除了 collect 之外，还有其他常见的末端操作符，它们大体分为两类：

集合类型转换操作符，包括 toList toSet
聚合操作符，包括将 flow 规约到单值的 reduce、fold 等操作；还有获得单个元素的操作符，包括 sing、singleOrNull、first 等。

实际上，识别是否为末端操作符，还有一个简单方法：由于 Flow 的消费端一定需要运行在协程中，因此末端操作符都是挂起函数。

分离 Flow 的消费和触发

我们除了可以在 collect 处消费 Flow 的元素以外，还可以通过 onEach 来做到这一点。这样消费的具体操作就不需要与末端操作符放到一起， collect 函数可以放到其他任意位置调用，例如代码如下所示。

// 分类 Flow 的消费和触发
fun createFlow() = flow<Int> {
    (1..3).forEach {
        emit(it)
        delay(100)
    }
}.onEach { println(it) }

fun main() {
    GlobalScope.launch {
        createFlow().collect()
    }
}

由此，我们又可以衍生出一种新的消费 Flow 的写法，代码如下所示。

fun main() {
    // 使用协程作用域直接触发 Flow
    createFlow().launchIn(GlobalScope)
}

其中， launchln 函数只接收 CoroutineScope 类型的参数。

Flow的取消

Flow 没有提供取消操作的方法，因为并不需要。
我们前面已经介绍了 Flow 的消费依赖于 collect 这样的末端操作符，而它们又必须在协程中调用，因此 Flow 的取消主要依赖于末端操作符所在的协程的状态。

也就是说， 要取消Flow只需要取消 flow.collect { } 所在的协程即可。
Flow 取消相关代码如下所示

// Flow 的取消
val job = GlobalScope.launch {
    val intFlow = flow {
        (1..3).forEach {
            delay(1000)
            emit(it)
        }
    }
    intFlow.collect {
        println(it)
    }
}
delay(2500)
job.cancelAndJoin()

其他Flow的创建方式

我们已经知道了 flow{…} 这种形式的创建方式，不过在这当中无法随意切换调度器，这是因为 emit 函数不是线程安全的，代码如下所示是错误示例。

// 不能在 flow 中直接切换调度器
flow { // BAD ！！
    emit(1)
    withContext(Dispatchers.IO) {
       emit(2)
    }
}

想要在生成元素时切换调度器，就必须使用 channelFlow 函数来创建 Flow:

channelFlow {
    send(1)
    withContext(Dispatchers.IO) {
        send(2)
    }
}

或者：

val channel = Channel<Int>()
channel.consumeAsFlow()

此外，我们也可以通过集合框架来创建 Flow:

listOf(1, 2, 3, 4).asFlow()
setOf(1, 2, 3, 4).asFlow()
flowOf(1, 2, 3, 4)

Flow的背压

只要是响应式编程，就一定会有背压问题，先来看看背压究竟是什么。背压问题在生产者的生产速率高于消费者的处理速率的情况下出现。为了保证数据不丢失，我们也会考虑添加缓冲来缓解背压问题，如下代码所示

// 为 Flow 添加缓冲
flow{
    List(100) {
        emit(it)
    }
}.buffer()

我们也可以为 buffer 指定一个容量。不过，如果只是单纯地添加缓冲，而不是根本上决问题，就会造成数据积压。
出现背压问题的根本原因是生产和消费速率不匹配，此时除可直接优化消费者的性能以外，还可以采取一些取舍手段。

第一种是 conflate 。与 Channel 的 Conflate 模式一致，新数据会覆盖老数据，例如下代码所示

// 使用 conflate 解决背压问题
flow{
    List(100) {
        emit(it)
    }
}.conflate()
.collect() {
    println("colleting $it")
    delay(100)
    println("$it collected")
}

我们快速发送了 100 个元素，最后接收到的只有 2 个，当然这个结果不一定每次都一样：

第二种是 collectLatest 顾名思义，其只处理最新的数据这看上去似乎与 conflate 没有区别，其实区别很大：collectLatest 并不会直接用新数据覆盖老数据，而是每一个数据都会被处理，只不过如果前一个还没被处理完后一个就来了的话，处理前一个数据的逻辑就会被取消。

// 使用 collectLatest 解决背压问题
suspend fun backPressure(){
    flow {
        emit(1)
        delay(50)
        emit(2)
    }.collectLatest { value ->
        println("Collecting $value")
        delay(100) // Emulate work
        println("$value collected")
    }
}

运行结果如下：

上面的例子collectLatest当中100毫秒之后只能接受到2，因为延时100的过程中发送2的时候会把1取消掉。
除了collectLatest 之外，还有 mapLatest、flatMapLatest 等，因为作用类似，故不再重复。

Flow的Map变换

flow{
    List(100) {
        emit(it)
    }
}.map {
    it * 2
}

也可以映射成其他Flow:

flow{
    List(5) {
        emit(it)
    }
}.map {
    flow{
        List(it) { emit(it) }
    }
}

实际上我们得到的是一个数据类型为 Flow 的 Flow, 如果希望将它们拼接起来，可以使用 flattenConcat, 如下代码所示：

flow{
    List(5) {
        emit(it)
    }
}.map {
    flow{
        List(it) { emit(it) }
    }
}.flattenConcat()
    .collect {
    println(it)
}

在拼接的操作中， flattenConcat 是按顺序拼接的，结果的顺序仍然是生产时的顺序。此外，我们还可以使用 flattenMerge 行会并发拼接，但得到的结果不会保证顺序与生产是一致。

Select 多路复用

在 UNIX 的 IO 多路复用中，我们应该都接触过 select, 其实在协程中，select 的作用也与在 UNIX 中类似。

复用多个await

我们前面已经接触过很多挂起函数，如果有这样一个场景，两个 API 分别从网络和本地缓存获取数据，期望哪个先返回就先用哪个做展示，实现代码如下代码所示。

// 网络获取用户信息
fun CoroutineScope.getUserFromApi(name: String) = async(Dispatchers.IO) {
    githubApi.getUserSuspend(name)
}

// 本地获取用户信息
fun CoroutineScope.getUserFromLocal(name: String) = async(Dispatchers.IO) {
    File(localDir, name).takeIf { it.exists()}
        ?.readText()
        ?.let {
            gson.fromGson(it, Uset::class.java)
        }
}

不管先调用哪个 API, 返回的 Deferred 的 await 都会被挂起，最终得到的结果可能并不是最先返回的，这不符合预期。当然，我们也可以启动两个协程来分别调用 await，不过这样会将问题复杂化。

接下来我们用 select 来解决这个问题，具体代码如下代码所示。

// 使用 select 复用 await
GlobalScope.launch {
    val name ="xxx"
    val localDeffered = getUserFromLocal(name)
    val remoteDeffered = getUserFromApi(name)
    
    val useResponse = select<Response<User?>> {
        localDeffered.onAwait{ Response(it, true) }
        remoteDeffered.onAwait{ Response(it, false) }
    }
    ...
}.join()

可以看到，我们没有直接调用 await, 是调用了 onAwait 在 select 中注册了回调，select 总是会立即调用最先返回的事件的回调。如图 6-6 示，假设 localDeferred.onAwait 先返回，那么 userResponse 的值就是 Response(it, true) ，由于我们的本地缓存可能不存在，因此 select 的结果类型是 Response<User?＞。

对千这个案例，如果先返回的是本地缓存，那么我们还需要获取网络结果来展示最终结果，完整代码如下所示。

val localDir = File("localCache").also { it.mkdirs() }
val gson = Gson()
// 网络获取用户信息
fun CoroutineScope.getUserFromApi(login: String) = async(Dispatchers.IO){
    gitHubServiceApi.getUserSuspend(login)
}
// 本地获取用户信息
fun CoroutineScope.getUserFromLocal(login:String) = async(Dispatchers.IO){
    File(localDir, login).takeIf { it.exists() }?.readText()?.let { gson.fromJson(it, User::class.java) }
}
fun cacheUser(login: String, user: User){
    File(localDir, login).writeText(gson.toJson(user))
}
data class Response<T>(val value: T, val isLocal: Boolean)

suspend fun main() {
    val login = "test"
    GlobalScope.launch {
        val localDeferred = getUserFromLocal(login)
        val remoteDeferred = getUserFromApi(login)

        //select选择优先返回的结果
        val userResponse = select<Response<User?>> {
            localDeferred.onAwait { Response(it, true) }
            remoteDeferred.onAwait { Response(it, false) }
        }

        userResponse.value?.let { println(it) } //获取结果显示 输出
        //如果是本地的结果，重新请求，并缓存本地
        userResponse.isLocal.takeIf { it }?.let {
            val userFromApi = remoteDeferred.await()
            cacheUser(login, userFromApi)
           println(userFromApi)
        }
    }.join()
}

复用多个channel

val channels = List(10) {Channel<Int>()}
GlobalScope.launch {
     delay(100)
     channels[Random.nextInt(10)].send(200)
}
val result = select<Int?> {
     channels.forEach { channel ->
        channel.onReceive { it }
     }
}
println(result)

对于 onRceceive, 如果 Channel 被关闭， select 会直接抛出异常 ；如果不希望抛出异常，可以使用 onReceiveCatching。

SelectClause

我们怎么知道哪些事件可以被 select 呢？其实所有能够被 select 的事件都是 SelectClauseN 类型，包括：

SelectClause0：对应事件没有返回值，例如 join 没有返回值，那么 onJoin 就是 SelectClauseN 类型使用时， onJoin 参数是个无参函数，如下所示。

// 复用无参数的join
select<Unit> {
    job.onJoin(){ println("Join resumed!")}
}

SelectClause1: 对应事件有返回值，前面的 onAwait 和 onReceive 都是此类情况
SelectClause2：对应事件有返回值，此外还需要一个额外的参数，例如 channel.onSend 有两个参数，第一个是 Channel 数据类型的值，表示即将发送的值；第二个是发送成功时的回调参数。相关代码如下所示。

// 复用两个参数的send
List(100){ element ->
    select<Unit> {
        channels.forEach { channel ->
            channel.onSend(element) { sentChannel ->
               println("sent on $sentChannel")
            }
        }
    }
}

onSend 的第个参数的 sentChannel 表示数据成功发送到的 Channel 对象。因此，如果大家想要确认挂起函数是否支持 select 只需要查看其是否存在对应的 SelectC!auseN 类型可回调即可。

另外官方的源码注释中也为我们列出了目前支持Select语法的挂起函数：
在这里插入图片描述

使用 Flow 实现多路复用

在这里插入图片描述

在代码清单 6-59 中，① 处建了由两个函数引用组成的 List; ② 处调用这两个函数得到deferred; ③ 处比较关键，对于每一个 deferred 我们创建一个单独的 Flow, 并在 Flow 内部发送 deferred. await( ）返回的结果，即返回的 User 象。现在有了两个 Flow 实例，我们需要将它们整合成一个 Flow 进行处理，此时调用 merge 函数即可，如图 6-7 所示。

同样， channel 读取复用的场景也可以使用 Flow 来完成：

// 使用 Flow 实现对 Channel 的复用
val channels = List(10) {Channel<Int>()}
...
val result = channels.map {
    it.consumeAsFlow()
}
.merge()
.first()

这比使用 select 实现的版本看上去要更简洁明了，每个 channel 都通过 consumeAsFlow 函数被映射成 Flow 再组合成 Flow, 取第一个元素。

协程中的线程安全问题

由于协程是运行在线程框架之上的，协程的异步回调依赖于其调度器的具体实现，而调度器的实现一般是基于线程的封装，所以在协程中依然存在线程安全问题（除非你只在单线程中使用它）。

不安全的并发访问

我们看一个简单的并发计数问题，代码如下所示

// 不安全的计数
suspend fun main() {
    var count = 0
    List(1000) {
        GlobalScope.launch {
            count++
        }
    }.joinAll()
    println("count: $count")
}

输出结果：

count：992

运行在 Java 平台上，默认启动的协程会被调度到 Default 这个基于线程池的调度器上，因此 count++是不安全的，最终的结果也证实了这一点。

不安全的原因主要有以下两点：

count++不是原子操作。
count 的修改不会立即刷新到主存，导致读写不一致。

解决这个问题我们都有丰富的经验，例如将 count 声明为原子类型，确保自增操作为原子操作，代码如下所示

// 确保修改的原子性
suspend fun main() {
    val count = AtomicInteger(0)
    List(1000) {
        GlobalScope.launch {
            count.getAndIncrement()
        }
    }.joinAll()
    println("count: ${count.get()}")
}

当然，直接粗暴地加锁也是一种思路，虽然我们都知道这不是一个好的解决方法。
使用synchronized加锁的代码如下：

suspend fun main() {
    val lock = "lock"  // 需要保证锁的是同一个对象
    var count = 0
    List(1000) {
        GlobalScope.launch {
            synchronized(lock) {
                count++
            }
        }
    }.joinAll()
    println("count: $count")
}

使用ReentrantLock加锁的代码如下：

suspend fun main() {
    var count = 0
    val lock = ReentrantLock()
    List(1000) {
        GlobalScope.launch {
            try {
                lock.lock()
                count++
            } finally {
                lock.unlock()
            }
        }
    }.joinAll()
    println("count: $count")
}

这跟在Java世界里使用加锁的方式没有区别。

协程的并发工具

除了我们在线程中常用的解决并发问题的手段之外，协程框架也提供了一些并发安全的工具，包括：

Channel: 并发安全的消息通道，我们已经非常熟悉
Mutex：轻量级锁，它的 lock 和 unlock 语义上与线程锁比较类似，之所以轻量是因为它在获取不到锁时不会阻塞线程而只是挂起等待锁的释放，代码如下所示

// Mutex使用示例
suspend fun main() {
    var count = 0
    val mutex = Mutex()
    List(1000) {
        GlobalScope.launch {
            mutex.withLock {
                count++
            }
        }
    }.joinAll()
    println("count: $count")
}

Semaphore：轻量级信号量，信号量可以有多个，协程在获取到信号量后即可执行并发操作。当 Semaphore 的参数为 1 时，效果等价于 Mutex，相关示例代码如下

//  Semaphore使用示例
suspend fun testSync() {
    var count = 0
    val semaphore = Semaphore(1)
    List(1000) {
        GlobalScope.launch {
            semaphore.withPermit {
                count++
            }
        }
    }.joinAll()
    println("count: $count")
}

与线程相比，协程的 API 在需要等待时挂起即可，因而显得更加轻量，加上它更具表现力的异步能力，只要使用得当，就可以用更少的资源实现更复杂的逻辑。

避免访问外部可变状态

我们前面一直在探讨如何正面解决线程安全的问题，实际上多数时候我们并不需要这么做。我们完全可以想办法规避因可变状态的共享而引发的安全问题，上述计数程序出现问题的根源是启动了多个协程且访问一个公共的变量 count，如果我们能避免在协程中访问可变的外部状态，就基本上不用担心并发安全的问题。

如果我们编写函数时要求它不得访问外部状态，只能基于参数做运算，通过返回值提供运算结果，这样的函数不论何时何地调用，只要传入的参数相同，结果就保持不变，因此它就是可靠的，这样的函数也被称为 纯函数。我们在设计基于协程的逻辑时，应当尽可能地编写纯函数，以降低程序出错的风险。

前面计数的例子的目的是在协程中确定数值的增量，那么我们完全可以改造成如下代码来满足需求

suspend fun main() { 
    val count = 0
    val result = count + List(1000) {
        GlobalScope.async { 1 }
    }.sumOf {
        it.await()
    }
    println("result: $result")
}