channel 进阶

前言

从 『Channel 是什么？』 中，我们已经清楚 channel 的基本使用以及其参数说明，下面，我们来继续学习它的更深入一点的知识。

迭代

在之前，我们都是通过 channel.receive() 一个个获取值，但是，其实还有一种方式，那就通过 iterator() 获得 ChannelIterator，然后进行遍历进行输出即可：

        val channel = Channel(3)

        GlobalScope.launch {
            launch {
                channel.send(1)
                channel.send(2)
                channel.send(3)
            }
            launch {
                val iterator = channel.iterator()
                while (iterator.hasNext()){
                    println("获取值：${iterator.next()}")
                }
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

日志输出：

I/System.out: 获取值：1
I/System.out: 获取值：2
I/System.out: 获取值：3
1
2
3

produce 和 actor

虽然 channel 能够很方便的提供发送和接收的功能，但是，对于单一职责而言，这并不符合，这是因为虽然我们使用封装了 send() 方式，但是由于使用者使用 receive() 的时候，也持有 channel，那就代表使用者也能够 send() 。

为了解决这种情况，我们使用 produce 代替生产者，使用 actor 代替消费者。

produce

        GlobalScope.launch {
            val produce = produce {
                (1..3).forEach { value ->
                    send(value)
                }
            }
            val iterator = produce.iterator()
            while (iterator.hasNext()){
                println("获取值：${iterator.next()}")
            }
        }
1
2
3
4
5
6
7
8
9
10
11

日志输出：

I/System.out: 获取值：1
I/System.out: 获取值：2
I/System.out: 获取值：3
1
2
3

actor

        GlobalScope.launch {
            val actor = actor {
                val iterator = iterator()
                while (iterator.hasNext()){
                    println("获取值：${iterator.next()}")
                }
            }

            (1..3).forEach { value ->
                actor.send(value)
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12

日志输出：

I/System.out: 获取值：1
I/System.out: 获取值：2
I/System.out: 获取值：3
1
2
3

channel 的关闭

cancel() 和 close() 都可以关闭 channel，不过的话，两个处理有些不太一样。

cancel()

在调用完 cancel() 后，若再使用 send() 或 receive() 就会默认抛出 CancellationException 异常，不过由于该异常会被协程静默处理掉了，所以不会引起 App 的崩溃。

        val channel = Channel()
        GlobalScope.launch {
            try {
                channel.cancel()
                channel.receive()
            }catch (e: Exception){
                println("捕获到异常: $e")
            }
        }
1
2
3
4
5
6
7
8
9

I/System.out: 捕获到异常: java.util.concurrent.CancellationException: RendezvousChannel was cancelled
1

close()

在调用完 close() 后，若再使用 send() ，就会抛出 ClosedSendChannelException，若再使用receive() ，就会默认抛出 ClosedReceiveChannelException 异常，要特别注意，这两个异常会引起 App 的崩溃，需要适时使用，若不知道该用哪个，就优使用 cancel()。

        val channel = Channel()
        GlobalScope.launch {
            channel.close()
            try {
                channel.send(1)
            }catch (e: Exception){
                println("捕获到 send() 异常: $e")
            }
            try {
                channel.receive()
            }catch (e: Exception){
                println("捕获到 receive() 异常: $e")
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

I/System.out: 捕获到 send() 异常: kotlinx.coroutines.channels.ClosedSendChannelException: Channel was closed
I/System.out: 捕获到 receive() 异常: kotlinx.coroutines.channels.ClosedReceiveChannelException: Channel was closed
1
2

特别说明

为了避免由于 channel 的关闭引发的异常，所以，我们在 send() 前可以进行 channel.isClosedForSend 判断，在 receive() 前进行 channel.isClosedForReceive 的判断。

        GlobalScope.launch {
            channel.close()
            if (channel.isClosedForSend){
                println("无法发送消息，channel 被关闭了")
            }else{
                channel.send(1)
            }
            
            if (channel.isClosedForReceive){
                println("无法接收消息，channel 被关闭了")
            }else{
                channel.receive()
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

I/System.out: 无法发送消息，channel 被关闭了
I/System.out: 无法接收消息，channel 被关闭了
1
2