Kotlin协程快速进阶

你们元旦快乐，去年（几天前）写了篇Kotlin协程快速入门，简单介绍了下协程的一些基本概念，今天来介绍下一些其余重要的知识点。

Channel

在协程里面开启另外一个协程是很方便的，但若是想在它们之间传递消息，或者说协程间通讯该怎么作呢？Channel（通道）就能够用做在协程之间简单的发送接收数据：

fun main() = runBlocking {
        val channel = Channel<String>()
        launch {
                channel.send("apple")
        }
        println("I like ${channel.receive()}")
    }
复制代码

这种作法是很像消费者与生产者模式。生产者一方生成并发送必定量的数据放到缓冲区中，与此同时，消费者也在缓冲区消耗这些数据。这一点经过它所继承的接口定义也能很好地体现：

public interface Channel<E> : SendChannel<E>, ReceiveChannel<E> {
  public companion object Factory {
       
        public const val UNLIMITED = Int.MAX_VALUE

        public const val RENDEZVOUS = 0

        public const val CONFLATED = -1
    }
}
复制代码

通道缓冲区 通道是一个接口，根据缓冲区容量不一样，有四种不一样的具体实现。

public fun <E> Channel(capacity: Int = RENDEZVOUS): Channel<E> =
    when (capacity) {
        RENDEZVOUS -> RendezvousChannel()
        UNLIMITED -> LinkedListChannel()
        CONFLATED -> ConflatedChannel()
        else -> ArrayChannel(capacity)
    }
复制代码

Channel的缓冲区默认是0个，当有信息send进去后，协程就会被挂起，只有被调receive后才会继续执行。若是容量大于0，当达到容量最大值时也一样会被挂起：

fun main() = runBlocking {
        val channel = Channel<Int>(2)
        launch {
            for (x in 1..5) {
                channel.send(x * x)
                println("send $x")
            }
        }
        delay(200L)
        repeat(2) { println("receive ${channel.receive()}") }
复制代码

结果以下，在发送两个数据后，只有收到一个数据后才会继续发送：

2019-01-01 19:01:11.176 30809-30809/com.renny.kotlin I/System.out: send 1
2019-01-01 19:01:11.176 30809-30809/com.renny.kotlin I/System.out: send 2
2019-01-01 19:01:11.377 30809-30809/com.renny.kotlin I/System.out: receive 1
2019-01-01 19:01:11.377 30809-30809/com.renny.kotlin I/System.out: receive 4
2019-01-01 19:01:11.377 30809-30809/com.renny.kotlin I/System.out: send 3
2019-01-01 19:01:11.377 30809-30809/com.renny.kotlin I/System.out: send 4
复制代码

以上就是Channel的基本用法了，看到这，熟悉Java并发编程的同窗很容易联想到阻塞队列，它们的做用是很类似的。Channel实现的阻塞队列并非真正的阻塞，而是协程被挂起，而且它是能够被关闭的。

Channel详解

上面说道Channel继承了SendChannel和ReceiveChannel，它自己没有实现逻辑，因此咱们来看下这两个接口的一些重要方法：

public fun offer(element: E): Boolean
复制代码

这也是发送消息的方法，不过和send不一样，它有返回值，在Channel缓冲区容量满了的时候不会挂起而是直接返回false。

public fun close(cause: Throwable? = null): Boolean
复制代码

关闭通道，关闭通道后再调用send或者offer会抛出异常。在发送方能够用isClosedForSend来判断通道是否关闭。对应的, 还有isClosedForReceive，但它会在全部以前发送的元素收到以后才返回 "true"。

public fun poll(): E?
复制代码

和offer对应，从缓冲区取不到消息会返回空，而不是像receive同样挂起协程。

public fun cancel(): Unit
复制代码

会取消接受消息并移除缓冲区的全部元素，所以isClosedForReceive也会当即返回"true"。

public operator fun iterator(): ChannelIterator<E>
复制代码

经过返回一个迭代器来接受缓冲区的消息，其实直接用for循环也是能够的（Channel并非一个集合，多是对协程的特殊支持吧）：

fun main() = runBlocking {
        val channel = Channel<Int>()
        launch {
            for (x in 1..5) channel.send(x * x)
            channel.close()
        }
        for (y in channel) println(y)
        println("Done!")
    }
复制代码

Channel进阶

事件的合并

再回到最初，缓冲区容量定义,大于等于0的值都很好理解，但Channel.CONFLATED = -1是什么鬼？ 咱们来改造下上面的demo：

fun main() = runBlocking {
        val channel = Channel<Int>(Channel.CONFLATED)
        launch {
            for (x in 1..5) {
                channel.send(x * x)
                println("send $x")
            }
        }
        delay(200L)
        repeat(2) { println("receive ${channel.receive()}") }
    }
复制代码

输出以下：

2019-01-01 20:10:29.922 1314-1314/com.renny.kotlin I/System.out: send 1
2019-01-01 20:10:29.922 1314-1314/com.renny.kotlin I/System.out: send 2
2019-01-01 20:10:29.927 1314-1314/com.renny.kotlin I/System.out: send 3
2019-01-01 20:10:29.927 1314-1314/com.renny.kotlin I/System.out: send 4
2019-01-01 20:10:29.928 1314-1314/com.renny.kotlin I/System.out: send 5
2019-01-01 20:10:30.117 1314-1314/com.renny.kotlin I/System.out: receive 25
复制代码

send方法并无被挂起，但咱们只收到了一个消息。事实上，定义为Channel.CONFLATED时，缓冲区的的容量也是1，但当容量已经有消息，但又有新消息来的的时候，它会用新消息来替代当前的消息。因此根据这个特性，接收方老是能接收到最新的消息。具体有啥用嘛？好比点击一次按钮触发一次动画，在动画播放期间的点击事件都将被合并成一次，当动画结束后，又会开始最新点击的动画，之间的点击都被略掉了。

扩展

上面发送和接受代码写的多少有些繁琐，官方还提供了扩展方法produce和consumeEach，咱们来该写下例子，不须要手动再开启发送消息一方的协程了：

fun main() = runBlocking {
        val squares = produce {
            for (x in 1..5) send(x * x)
        }
        squares.consumeEach { println(it) }
        println("Done!")
    }
复制代码

async/await

async 异步， await 等待 ，这两个方法是协程为了更好解决异步任务而推出的，熟悉JS、C#等语言的人对这两个方法确定很熟悉，用法也是差很少的。

fun main() = runBlocking{
        var time = measureTimeMillis {
            val one = doSomethingUsefulOne()
            val two = doSomethingUsefulTwo()
            println("The answer is ${one + two}")
        }
        println("Sync completed in $time ms")

         time = measureTimeMillis {
            val one = async { doSomethingUsefulOne() }
            val two = async { doSomethingUsefulTwo() }
            println("The answer is ${one.await() + two.await()}")
        }
        println("Async completed in $time ms")
    }

    suspend fun doSomethingUsefulOne(): Int {
        delay(1000L)
        return 13
    }

    suspend fun doSomethingUsefulTwo(): Int {
        delay(1000L)
        return 29
    }
复制代码

结果是同样的，但耗时却差了一半。咱们就像调用同步任务同样启用异步，不得不说比java原生实现优雅多了

2019-01-01 20:52:15.482 3520-3520/com.renny.kotlin I/System.out: The answer is 42
2019-01-01 20:52:15.483 3520-3520/com.renny.kotlin I/System.out: Sync completed in 2006 ms
2019-01-01 20:52:16.489 3520-3520/com.renny.kotlin I/System.out: The answer is 42
2019-01-01 20:52:16.489 3520-3520/com.renny.kotlin I/System.out: Async completed in 1006 ms
复制代码

async是一个扩展方法，在里面启动了一个子协程，看下定义：

public fun <T> CoroutineScope.async(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> T
): Deferred<T> {
    val newContext = newCoroutineContext(context)
    val coroutine = if (start.isLazy)
        LazyDeferredCoroutine(newContext, block) else
        DeferredCoroutine<T>(newContext, active = true)
    coroutine.start(start, coroutine, block)
    return coroutine
}
复制代码

返回的也再也不是Job对象，而是Deferred，这二者大概像就是runnable和callable的关系吧，无返回值和有返回值，其余都差很少。而await会挂起当前的协程，直到子协程代码结束并拿到返回结果，和join也相似。

小结

今天就介绍这么多啦，正如标题所说，这几篇文章的目的就是让你们一块儿更简单地和快速地对协程有个初步的了解。同RxJava同样，协程也是Kotlin为了更好地处理异步任务而推出的功能库，如何将其用在网络请求、数据库、文件IO等方面，让代码变得更简洁更优雅才是最终目的。