稀有猿诉

十年磨一剑,历炼出锋芒,说话千百句,不如码二行。

Kotlin进阶之协程从专家到出家

协程Coroutine是一种轻量级的实现异步的方式,是代码执行的控制框架,是最新式的发编程范式。它在使用上大大简化了以往多线程并发带来的种种麻烦(如状态同步和锁),最大的优势就是它是纯编程语言层面实现的,不受制于操作系统,因此它不会阻塞。在前两篇基础之上,今天再来深入的理解一下各种概念,并脱去协程华丽的外衣一探它的本质。

本篇为协程三步曲中的终篇:

不过如此

协程是最新式的并发编程范式,所有文章都会大量的提及同步,异步,并发和并行等,需要先理解一下这些词语到底是什么意思。

什么是异步(Asynchronous)

在解释什么是协程之前,得先要理解什么是异步(Asynchronous)。异步也就是说程序是非同步的(Non synchronous),并不是按照顺序来执行的。这么说依然很绕,我们来一个简单的例子。比如有这样一个语句:

1
2
3
4
fun main(args: Array<String>) {
     println("world of Coroutine!") // 1
     print("Hello ...") // 2
}

这个代码的输出很明显是

1
2
world of Coroutine
Hello ...

函数的执行是从上到下按我们写的顺序执行的,这就是顺序执行的意思,虽然说编译器会做一些指令重排以期对字节码进行一些优化,但有一个前提就是它不会改变程序的正确性,比如后面语句如果需要前面表达的结果时,一定能保证它们的执行顺序。同步的意思是,执行一个子函数,子函数会占用CPU,直到它运行结束再返回到调用它的函数,继续运行并能把结果带回给调用者,这即是同步的意思。比如这里的println,println没有执行完时,后面的语句是不会执行的。

异步的意思是函数语句的执行并不是按照我们写的顺序来运行的。比如说,前面的函数,如何能输出”Hello …world of Coroutine” ?这就需要让代码实现异步,非顺序的执行。有多种方式,协程就可以实现异步:

1
2
3
4
5
6
7
8
fun main() = runBlocking { // this: CoroutineScope
    launch { // launch a new coroutine and continue
        delay(1000L) // non-blocking delay for 1 second (default time unit is ms)
        println("world of Coroutine!") // print after delay
    }
    print("Hello ...") // main coroutine continues while a previous one is delayed
}
// Hello ...world of Coroutine!

什么是并发(Concurrency)

并发concurrency)就是代码同时开始运行,并发,并驾齐驱同时出发,英文是concurrent programming或concurrency,但出发了后是并行还是串行,并不管或者说不是重点。

异步是并发的前提,如果无法异步,那就不可能让代码同时开始运行,自然也就无法并发。并发是为了提高代码的运行响应效率或者说提升性能。

扩展阅读 并发(Concurrency)与并行(Parallelism)

什么是并行(Parallelism)

并行则是代码同时运行,任何时间段,都在同时运行。并行往往涉及复杂的计算任务,或者为解决计算量超级大的任务,需要借助专门的系统工具,把任务分解为不相关的子任务,然后分别运行在不同的计算机上面。所以大多数场景下,并行是并行计算(Parallel computing)的简称。除非专门从事于并行计算相关的工作,否则日常的软件开发基本上不涉及并行,大家平时接触最多的还是并发编程。

并发与并行的区别可以看这里

并发编程简史

随着软件的复杂度越来越高,以及多核心CPU的普及,让真并发变成了可能,导致并发编程已经变成了软件的一个非常基本的要求。现在,只要是软件,从Web到客户端,从商业软件到智能手机的App,甚至命令行工具也都需要并发,以提升性能和响应效率。从进程到线程再到协程,这其实是一部并发编程的历史。

进程(Processes)

是操作系统的概念,进程process)就是一个运行中的程序状态总和,包括代码和其所持有的资源。进程是操作系统管理运行中的程序的基本单元,未运行的程序不是进程,它只是一坨文件,当然了它可能会是其他进程持有的资源。

进程是最为原始的并发编程方式,不过严重依赖于操作系统,比如说创建,通信和同步都需要系统调用system call,如fork(),waitpid(),pipe(), socket()以及像semphore。进程虽然可以实现并发,但难以控制,可操控性太差,一般只作为粒度特别大的并发任务时,比如说上下游关系,不需要来回交互时,一个文本编辑器,需要打开一个HTTP超链接,可以启动浏览器,然后就跳转到浏览器了,啥时回来,回不回来都不重要,并不影响文本编辑器。

扩展阅读 如何理解:程序、进程、线程、并发、并行、高并发?

线程(Threads)

线程thread)是更进一步的代码执行控制模式,它是轻量级的进程。每个进程至少由一个线程组成。线程是运行中的代码,它有自己的调用栈,是操作系统调度代码运行的最小单元。线程要依赖于操作系统的支持,比如pthread。每个线程要真实的运行在CPU上面,并且会独占CPU。现代的CPU都是有多个执行核心的,每个核心都可以跑一个线程,所以现代的CPU可以真实的让线程并行的运行。

线程是操作系统调度代码运行的最小单元,每个线程会独点一个CPU核心。当线程被调度到时,就会占有CPU,直到它运行完,或者被阻塞,一旦线程阻塞了,它就失去了CPU控制权,它的代码自然也就停止运行了,直到再次被操作系统调度得到CPU控制权,从阻塞状态再次回到运行状态。

从线程出现,一直到现在线程都是最为主要的并发编程方式,现在的软件效率都很高,服务器的高并发等等都是依赖于线程实现的。线程虽然较进程轻量,但仍依赖于操作系统,先是要操作系统支持线程,其调度也是依赖于操作系统的。线程也会占用不少的资源,它需要有自己的调用栈和CPU的上下文环境,只不过线程与线程能共享进程的资源而已,进程是程序运行和资源总和,运行其实都是由线程在控制。

协程(Coroutines)

技术在不断的演进,新式的并发实现方式也在不断的涌现,新一代的技术总是能解决上一代的问题,从而慢慢取而代之,就如线程之于进程。而新的挑战者,便是协程coroutine

需要特别注意的是协程与线程没有关系,它是代码执行的操作框架,是实现异步和并发的最新的方式,它是让多个函数更好的协作以实现异步和并发。它是纯的编程语言层面的框架,不依赖于操作系统,因此它更为轻量,完全受开发者控制,与线程也没有关系,只会挂起,也就是某个协程停止执行,但不会阻塞线程,线程仍能去执行其他协程。相互协作的子例程,即为协程

线程是操作系统调度代码运行的最小单元,所有的代码想要运行,必然要跑在某个线程里面,协程是代码运行的操作框架,所以协程自然也要跑在线程里面。或者说协程是更为轻量的线程,一个线程可以运行多个协程。

自上而下的来看,操作系统包含多个进程,每个进程有多个线程,而每个线程有多个协程。进程与进程之间共享操作系统提供的资源,如文件系统;进程分为程序的运行和资源,程序的运行就是线程(栈),资源就是堆内存;线程是代码的运行,其有独立的栈来存储自己的数据,包括执行中的代码和产生的本地数据,都在栈帧上面,此栈是操作系统中的代码栈,代表着CPU的运行,因此线程仍需操作系统支持,进程中的不同线程共享进程的堆内存,可以共享堆数据,当然也带来了麻烦;协程用来操作线程中代码运行的框架,是粒度更为小的代码的运行,但它并没有真实的栈,因此协程只是运行中的代码,并没有自己的数据,当然也可以共享堆内存。

脱去外衣看本质

我们来了解一下协程的基本原理,以更深刻的理解协程。

本质是状态机+跳转

前面说过协程是代码运行的控制框架,一个协程可以挂起,其他协程得以继续执行,如果不是因为操作系统在调度,在编程语言层面,这其实就是从一处代码跳转到了另一处代码。因此,协程的核心原理是代码的跳转。如何跳,跳到哪,又是由状态机确定的(或者叫做『事件驱动模式』)。看大Python就明白了,Python中协程其实叫async I/O

想像有一个巨大的switch/when语句,遇到某个条件时会执行函数A(协程A),它改变条件后break(挂起了),条件变了会进入不同的分支,执行另一个函数B(协程B),协程B仍可以改变条件,再回到原来的分支去执行函数A(协程A继续执行了)。

除了switch/when,还可以用如goto/longjmp之类的语句实现跳转。

有栈(Stackful)和无栈(Stackless)

有两种协程的实现方式,一种称之为有栈的Stackful,另一种则是无栈Stackless。

  • Stackfullness的意思是指可以在当前代码的调用栈(可能是非协程)中挂起,恢复的时候能在挂起的地方继续执行,所有的函数栈都还在。
  • Stackless的意思是挂起只能发生在协程中。

什么意思呢,我们来理解一下,如果可挂起的函数只能运行在协程中或者被其他可挂起的函数调用,那么就是stackless的,意思就是说可挂起的函数必须是协程里面的顶层函数,那么协程启动之前的栈对于协程来说都是不可见的。就比如Kotlin就是stackless的,因为suspend函数只能由协程调用或者另一个suspend调,它不能在常规的调用栈中,必须为它启动一个协程。

Stackful需要每个协程开辟额外的空间来保存直到挂起点的stack,但它可以实现一些非常复杂的功能,相当于supsend与非suspend可以混着用,随时挂起,继续执行的时候原先的函数栈都还在,当然了这么实现起来代价肯定也是要大一些的。而stackless虽然限制多一些,只能由协程调用一个supsend函数,但效率高啊,实现起来也稍微简单。

扩展阅读:

CPS(Continuation Passing Style)

CPS(Continuation Passing Style)续体传递风格,是一种函数式的编程风格,函数并不直接返回结果,而是接收一个代码块作为最后一个参数。这个代码块会在函数要执行完被调用执行,用以处理函数结果。其实这个代码块就是回调函数,称之为续体(Continuation),它会决定程序接下来的行为。整个程序就通过一个一个的Continuation拼接在一起。

如果以同步式的写法,函数处理foo的返回值,可这样写:

1
2
3
4
val a = foo()
if (a == 200) {
   bar()
}

如果改成CPS式,就是酱婶儿的:

1
2
3
4
5
6
7
8
9
10
foo() { a->
   if (a == 200) {
      bar()
   }
}

fun foo(block: (Int)->Unit) {
   val a = compute() // compute to get value
   block(a) // execute the block at the end
}

只有到了挂起点才会挂起

需要注意的是,协程的代码并不是随时挂起,只有遇到了挂起点才会挂起。什么是挂起点呢?就是执行到了被suspend修饰的函数时,就会被挂起。协程是编程语言自己实现的代码执行控制框架,所以对于协程的实现者来说决定在哪里挂起是可控的。

扩展阅读:

协程是如何实现的

下面看一下协程在Kotlin的内部机制,以了解一下Kotlin是如何实现协程的。这里并不是为了深究它的好与坏,了解底层的知识都是为了更好的运用上层知识。另外要注意,这部分为了厘清概念用的都是伪码,并不是完全可运行的代码。

Kotlin协程的内部实现

借助CPS和状态机就可以实现协程,在Kotlin中就是这么做的。我们通过一个简单的例子就能把它的原理讲清楚。

前面讲了CPS是把函数调用后续操作封装成为一个代码块,传给调用的函数,其会在函数体最后执行此代码块。代码块就是针对 结果操作的一坨代码,为了更加通用,可以声明一个接口,作为续体的类型。在Kotlin中,这便是接口Continuation

1
2
3
4
5
ublic interface Continuation<in T> {
    public val context: CoroutineContext
    public fun resume(value: T)
    public fun resumeWithException(exception: Throwable)
}

创建一个实现了此接口的对象就可以当作续体传递给另一个函数。当遇到被suspend修饰的函数时,Kotlin的编译器就会生成一个继体对象,此对象会持有suspend函数运行相关的状态,如内部函数的返回值,以及状态机的状态。同时生成一个状态机函数,状态机接收续体对象作为参数,根根据其状态标签,执行不同的内部suspend函数,并把续体继续当作参数传给其他的suspend函数,直到退出。

举个🌰,对于酱紫的suspend函数:

1
2
3
4
5
6
7
suspend fun getUserSummary(id: Int): UserSummary {
    logger.log("fetching summary of $id")
    val profile = fetchProfile(id) // suspending fun
    val age = calculateAge(profile.dateOfBirth)
    val terms = validateTerms(profile.country, age) // suspending fun
    return UserSummary(profile, age, terms)
}

里面会有两个挂起点,把函数体分成三部分,所以可以视为一个三个状态的状态机

1
2
3
4
5
6
7
8
9
10
suspend fun getUserSummary(id: Int): UserSummary {
    // label 0 -> first execution
    logger.log("fetching summary of $id")
    val profile = fetchProfile(id)
    // label 1 -> resuming
    val age = calculateAge(profile.dateOfBirth)
    val terms = validateTerms(profile.country, age)
    // label 2 -> resuming
    return UserSummary(profile, age, terms)
}

最终会生成续体对象,续体包含挂起点之后代码继续运行的所有必要参数,以及状态机函数,大概会是酱婶儿的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
private class GetUserSummarySM: Continuation<Any?> {
    var value: Any? = null
    var cont: Continuation<Any?>? = null
    val id: Int? = null
    var profile: Profile? = null
    var age: Int? = null
    var terms: Terms? = null

    override fun resume(data: Any?) {
        this.value = data
        getUserSummary(id, this)
    }
}

suspend fun getUserSummary(id: Int, cont: Continuation<Any?>) {
    val sm = cont as? GetUserSummarySM ?: GetUserSummarySM()
    when(sm.label) {
        0 -> { // Label 0 -> first execution
            sm.cont = cont
            logger.log("fetching summary of $id")
            sm.label = 1
            fetchProfile(id, sm)
            return
        }
        1 -> { // label 1 -> resuming
            sm.profile = sm.value as Profile
            sm.age = calculateAge(sm.profile!!.dateOfBirth)
            sm.label = 2
            validateTerms(sm.profile!!.country, sm.age!!, sm)
            return
        }
        2 -> { // label 2 -> resuming and terminating
            sm.terms = sm.value as Terms
            sm.cont!!.resume(UserSummary(sm.profile!!, sm.age!!, sm.terms!!))
        }
    }
}

这里的原理并不复杂,我们捋一捋就能弄懂了:状态机函数,就是一个大的when语句。初始状态是没有续体的,创建一个续体对象,并执行第一部分代码,直到挂起点,修改续体到状态2,然后执行suspending函数,此时便挂起了,也就是说其实函数已执行完了。当挂起结束后,继续运行时,就回调续体里面的resume,仔细看resume又会回到状态机,只不过这时状态已在挂起前改为2了,所以会执行第2部分的代码。同理修改状态为3,然后再次挂起,等继续时就会执行到第3部分代码,这时没有挂起点了,所以不再传入续体,整体协程结束了。

厘清后,发现协程的实现非常的巧妙:挂起点把函数分为不同部分,用状态区分,用when执行不同的部分;挂起前把状态改为下一部分,然后把状态机函数当成续体,传给挂起点的suspending函数,当挂起结束后执行续体,便又回到状态机,但会进入新的状态,执行余下部分的代码。就这样,一次执行一部分代码,直到没有挂起点了,不再传入续体了,原suspending函数也就结束了。

扩展阅读

其他编程语言的协程

比较值得学习就是Go语言的协程Goroutines,Go语言是比较早且比较完整的支持协程的编程语言,并且它的协程是stackful的。

扩展阅读:

以及大Python中的async I/O

扩展阅读:

测试Testing

这里主要指的是针对开发人猿的单元测试,而非测试同学的验收测试。

协程是一种并发编程范式,因此,测试与并发代码测试是一样的。无论哪种测试,本质都是一样的,把代码当成黑盒,给特定的输入,看是否能产生符合预期的输出。特定的输入通常需要Mock。

需要测试的是逻辑

一定要厘清,逻辑代码是需要测试的,要把重点放在逻辑的测试,啥是逻辑?其实就是条件控制语句和循环语句,所以要盯着条件和循环来写case。

比如说,网络请求,要把无网络,连接超时,服务器返回超时,返回码不对,返回数据格式不对,这些情况都Mock到,以测试逻辑是否有把所有的可能情况都考虑到了。

TestScope

Kotlin的协程库特意准备了一个专门用于测试的TestScope,这个scope的好处在于它使用的是虚拟时间。时间对于并发编程是特别重要的,比如经常需要等待多少时间间隔(协程相互协作的等待,或者等待服务器返回),以及异步任务的超时时间,在实际的运行中肯定是真实的去等。但对于测试代码来说,也这样真的去等,就有点难以接受了,单元测试的一个非常重要的要求就是测试必须要快,要能快速反馈结果。

为此,就有了TestScope,它用的是虚拟时间,可以理解为这个scope中delay的时间是一个虚拟的时间,它会很快的执行,相当于这里的时间变快了。因为它相当于是整体的时间都变快了,所以里面代码相对的时序并不受影响,所以不会影响原本应该有的时序。可以理解为仙界,仙界一天地界一年。

扩展阅读 Testing coroutines

并发安全性(Concurrent safety)

这里指的是并发程序中共享数据的并发安全性(Concurrent safety),而非数据隐私安全的那个安全(data security)。

线程有自己的代码,但却可以与其他线程共享数据,代码操作数据,看到的数据不一致,就产生了并发安全问题,并发安全问题是由多线程共享变量(shared mutable data)引发的,所以要想解决并发安全,要么别共享变量(无论是使用常量Immutable data,避免共享变量,用线程自己的变量ThreadLocal,用锁来保证共享变量的原子访问);要么别用多线程。

协程也是并发编程,如果协程切换了不同的运行线程,且又访问了共享数据,那就有可能有并发安全。注意并发安全问题(共享可变数据不一致)是由多线程引起的,如果没有切换线程,则没有问题,我们可以验证一下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
fun main() = runBlocking{
    var count = 0

    val jobs = mutableListOf<Job>()
    val timeCost = measureTimeMillis {
        repeat(1000) {
            val job = launch {
                delay(100)
                count++
            }
            jobs.add(job)
        }
        jobs.joinAll()
    }
    log("timeCost = $timeCost")
    log("count = $count")
}
//[main @coroutine#1] timeCost = 173
//[main @coroutine#1] count = 1000

这个示例中多个协程共享了一个变量,并且++并不是原子的,但结果是对的,说明没有发生数据不一致的问题,原因是没有为协程切换线程。我们切一下线程试试:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
fun main() = runBlocking{
    var count = 0

    val jobs = mutableListOf<Job>()
    val timeCost = measureTimeMillis {
        repeat(1000) {
            val job = launch(Dispatchers.Default) {
                delay(100)
                count++
            }
            jobs.add(job)
        }
        jobs.joinAll()
    }
    log("timeCost = $timeCost")
    log("count = $count")
}
//[main @coroutine#1] timeCost = 165
//[main @coroutine#1] count = 998

这回可以看到数据不一致的问题发生了,因为结果并不是预期的1000。

想要减少并发安全问题,可以多多使用常量(Immutable data),此外如果必须要共享变量,那么可以用协程专用的互斥锁。注意,线程专用的原子化数据结构如AtomicInteger等也是可以使用的,并且效率很高,但只能用于一些有限的数据结构如如基础类型或者集合;其他的线程同步方式如synchronized/ReentrantLock/CountDownLatch等会阻塞线程,会严重影响性能,毕竟它们是为多线程并发准备的工具,在协程中并不推荐使用。

轻量级锁

协程专用的互斥锁(Mutual exclusion)是Mutex,它是轻量级的锁,原因就是它只会挂起,而非阻塞。可以用它来保护对共享变量的操作,以解决并发一致性问题。它的使用方法与Java中的ReentrantLock是一样的,可以用lock/unlock式,或者try-lock-finally-unlock式,或者更为方便的扩展函数withLock {…},比如酱婶儿的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
fun main() = runBlocking{
    var count = 0
    val mutex = Mutex()

    val jobs = mutableListOf<Job>()
    val timeCost = measureTimeMillis {
        repeat(1000) {
            val job = launch(Dispatchers.Default) {
                delay(100)
                try {
                    mutex.lock()
                    count++
                } finally {
                    mutex.unlock()
                }
            }
            jobs.add(job)
        }
        jobs.joinAll()
    }
    log("timeCost = $timeCost")
    log("count = $count")
}
//[main @coroutine#1] timeCost = 193
//[main @coroutine#1] count = 1000

和用withLock,是一样一样的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
fun main() = runBlocking{
    var count = 0
    val mutex = Mutex()

    val jobs = mutableListOf<Job>()
    val timeCost = measureTimeMillis {
        repeat(1000) {
            val job = launch(Dispatchers.Default) {
                delay(100)
                mutex.withLock {
                    count++
                }
            }
            jobs.add(job)
        }
        jobs.joinAll()
    }
    log("timeCost = $timeCost")
    log("count = $count")
}
//[main @coroutine#1] timeCost = 191
//[main @coroutine#1] count = 1000

要保护共享变量的操作而非协程

需要特别注意,协程是异步代码,不能以同步思维来看待。我们用锁是来保护协程内部对共享变量的操作,防止数据不一致,因此要把锁放在共享变量操作的地方(critical section)。妄图大范围的对整个协程加锁是没有用的,比如酱紫:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
fun main() = runBlocking{
    var count = 0
    val mutex = Mutex()

    val jobs = mutableListOf<Job>()
    val timeCost = measureTimeMillis {
        repeat(1000) {
            mutex.withLock { // Do NOT do this, it does NOT work.
                val job = launch(Dispatchers.Default) {
                    delay(100)
                    count++
                }
                jobs.add(job)
            }
        }
        jobs.joinAll()
    }
    log("timeCost = $timeCost")
    log("count = $count")
}
//[main @coroutine#1] timeCost = 165
//[main @coroutine#1] count = 976

这里失效的原因与异常捕获是一样的,协程并不是同步代码(not synchronous),它是异步的(asynchronous),外围的try/catch和lock之类的操作对异步代码是没有效用的。

扩展阅读

书籍推荐

《七周七并发模型》 这本书是专门讲并发的,并且讲了多种语言的并发模式,是理解并发非常好的书籍,值得精读。

参考资料

Comments