Kotlin从入门到放弃（三）——协程

2022-01-17 15:28:06

引言

这篇主要是将以下kotlin里面的协程，当然这个概念已经随着kotlin的文档被广泛得知了，不过还是用大量代码记录一下吧

一、概念

Coroutine，翻译为协程，意思为各个子任务程协作运行。由此可以联想到Java常用的线程概念，java中的线程Thread最终启动的地方是JVM核心层，也就是说java的线程其实本质也是和硬件有关（这是当然的）。而多线程任务在并发的情况下会出现阻塞的情况，协程提供了挂起这种方法去避免阻塞线程并用更廉价更可控的操作替代线程阻塞。

二、入门

协程是由程序直接实现的，是一种轻量级线程，kotlin也为此提供了标准库和额外的实验库。标准库为kotlin.coroutines.experimental（写作时使用kotlin-1.20版本），可见仍然还是一个实验性功能。其实协程的例子能在网上找到很多，就不一一去分析了，这里以标准库实现的斐波那契方法为例讲解一下。

val fibonacciSeq = buildSequence {
        var a : Long = 0
        var b : Long = 1
        yield(1)           // 1
        while (true) {
            yield(a + b)   // 2
            val tmp = a + b
            a = b
            b = tmp
            print(tmp.toString() + " ") // 3
        }
    }
println(fibonacciSeq.take(10).toList()) // 4

上面的fibonacciSeq函数实现了斐波那契数列，但是不同于java语法写的方法这里多了一个buildSequence，buildSequence源码如下：

/**
 * Builds a [Sequence] lazily yielding values one by one.
 *
 * @see kotlin.sequences.generateSequence
 *
 * @sample samples.collections.Sequences.Building.buildSequenceYieldAll
 * @sample samples.collections.Sequences.Building.buildFibonacciSequence
 */
@SinceKotlin("1.1")
public fun <T> buildSequence(builderAction: suspend SequenceBuilder<T>.() -> Unit): Sequence<T> = Sequence { buildIterator(builderAction) }

从官方的注释中可以知道，这是创建了一个惰性的序列，也就是函数fibonacciSeq是一个创建无穷惰性的斐波那契数列。注释1处的yield(1)作用就是输出1，yield的作用可以参考Python语言的相当于return；同样注释2处也同理；注释3和4处均是输出，不同的是4处可以主动取出有限的数列，两者的输出作为对比去理解yield在此处的return作用。

上图可以明显看出注释3处少输出了一个元素，这是因为yield在第10次直接返回，跳出了while循环。构建这种无限序列（后面还会讲）是协程的一个主要特点，不过可能从这里很难看出它的概念（子任务），理解任务的调度可以引入kotlin提供的一个额外Coroutine库——kotlinx.coroutines.experimental，显而易见也是试验阶段。

三、实践

3.1 启动协程

使用kotlinx库就需要gradle了（当然其他的如Maven也是可以的，手动滑稽），版本选择和导入不再赘述，还是直接上代码。

    // 在Common线程池启动协程
    launch(CommonPool) { 
        delay(2000L)    // 1
        println("Hello")
    }
    println("World")
    Thread.sleep(3000L) // 2
    // 在主线程中启动协程
    runBlocking<Unit> {
        println("T0")
        launch(CommonPool) {
            println("T1")
            delay(3000L)
            println("T2 Hello")
        }
        println("T3 World")
        delay(5000L)
        println("T4")
    }

此处将两个函数写在一处好做对比，1处的delay函数（非阻塞）是一个suspend（挂起）函数，在这里的作用相当于2处的Thread.sleep（阻塞），但是delay必须在协程中或者挂起函数中使用，但是lauch是在CommonPool共享线程池中创建协程并不是主线程，所以不能使用，解决的办法就是使用runBlocking——桥接普通阻塞代码和挂起风格的非阻塞代码，即可以在里面启动协程，使用挂起函数和常用的阻塞方法。

3.2 取消协程

有启动自然就有取消操作，使用cancel函数，当然这里也是有需要注意的点。

runBlocking {
        val job = launch {
            repeat(1000) {
                i -> println("job sleeping $i ... CurrentThread: ${Thread.currentThread()}")
                delay(500L)
            }

        }

        val job1 = launch {
            var nextTime = 0L
            var i = 0
            while (i < 20) {
                var currentTime = System.currentTimeMillis();
                if (currentTime >= nextTime) {
                    println("job1 sleeping ${i++} ... CurrentThread: ${Thread.currentThread()}")
                    nextTime = currentTime + 500L
                }
            }
        }
        delay(1900L)
        println("Job is alive: ${job.isActive}; Job iscompleted: ${job.isCompleted}")
        println("Job1 is alive: ${job1.isActive}; Job1 iscompleted: ${job1.isCompleted}")
        val b1 = job.cancel()
        val c1 = job1.cancel()
        println("job cancel: $b1 and job1 cancel: $c1")
        delay(1300L)
        println("Job is alive: ${job.isActive}; Job iscompleted: ${job.isCompleted}")
        println("Job1 is alive: ${job1.isActive}; Job1 iscompleted: ${job1.isCompleted}")
        delay(30000L)
        val b2 = job.cancel()
        val c2 = job1.cancel()
        println("job cancel: $b2 and job1 cancel: $c2")
        println("Job is alive: ${job.isActive}; Job iscompleted: ${job.isCompleted}")
        println("Job1 is alive: ${job1.isActive}; Job1 iscompleted: ${job1.isCompleted}")
    }

同样是两个协程job和job1，job是一个可重复1000次但会被挂载的协程而job1是一个有时间间隔循环20次的协程，这里就不截取输出的结果了因为比较长。协程job由于使用了delay函数挂起，在调用了cancel之后协程实现了真正的停止；协程job1存在循环计算并没有挂起操作，即使调用了cancel，协程状态也变为停止，但是循环操作仍然在继续，这种情况下取消就会失效。那遇到第二种情况该怎么办嘞，联系到上面讲过的yield函数（注意这里的yield函数是kotlinx包中的和上面那个重名），写在适当的地方就可以了（话说还不如直接用delay）。

3.3 等待协程

之前的代码中实现的功能不同，但是有个共同的特点，那就是主线程主动挂起或者阻塞等待协程里面的代码执行，这在实际使用中肯定是不可取。kotlinx提供了join函数，可以使主线程等待协程执行完。

runBlocking {
    var c1 = launch(CommonPool) {
        delay(1000L)
        println("Coroutine 1")
    }
    var c2 = launch {
        delay(1000L)
        println("Coroutine 2")
    }
    c1.join() // 1
    c2.join() // 2
    println("the main")
}

join函数也是一个挂起函数源码解析放在之后吧，如果没有1和2处的代码，整个程序的运行结果只有“the main”，而现在的执行结果如下图：

上图的目的就是为了说明协程的执行也是无序的。

3.4 CommonPool线程池

官方的解释是这样的Represents common pool of shared threads as coroutine dispatcher for compute-intensive tasks，机器翻译一下就是将共享线程池作为一个协程来调度计算密集型任务。注释1处会尝试新建一个ForkJoinPool（一个可执行ForkJoinTask的ExcuteService，采用工作窃取算法：所有在池中的线程尝试去执行其他线程创建的子任务，这样很少有线程处于空闲状态，更加高效）；如果不可用，就是用Executors来创建一个普通的线程池，创建过程在注释3处。

    private fun createPool(): ExecutorService {
        val fjpClass = Try { Class.forName("java.util.concurrent.ForkJoinPool") }
            ?: return createPlainPool()   // 1
        if (!usePrivatePool) {
            Try { fjpClass.getMethod("commonPool")?.invoke(null) as? ExecutorService }
                ?.let { return it }
        }
        Try { fjpClass.getConstructor(Int::class.java).newInstance(defaultParallelism()) as? ExecutorService }
            ?. let { return it }
        return createPlainPool()  // 2
    }

    private fun createPlainPool(): ExecutorService {
        val threadId = AtomicInteger()
        return Executors.newFixedThreadPool(defaultParallelism()) {
            Thread(it, "CommonPool-worker-${threadId.incrementAndGet()}").apply { isDaemon = true }  // 3
        }
    }

    private fun defaultParallelism() = (Runtime.getRuntime().availableProcessors() - 1).coerceAtLeast(1)

四、小结

协程的概念和基本操作讲的差不多了，对于经常接触Thread线程突然去理解协程还是有点障碍的，不过官方提供了很好的试验库去入门。协程也不是一两篇文章就能讲清楚的，应该还会有下一篇继续介绍。

码农公寓

引言