kotlin最新协程教程——1.2.10
协程与线程的区别：
在高并发的场景下，多个协程可以共享一个或者多个线程，性能可能会要好一些。举个简单的例子，一台服务器有 1k 用户与之连接，如果我们采用类似于 Tomcat 的实现方式，一个用户开一个线程去处理请求，那么我们将要开 1k 个线程，这算是个不小的数目了；而我们如果使用协程，为每一个用户创建一个协程，考虑到同一时刻并不是所有用户都需要数据传输，因此我们并不需要同时处理所有用户的请求，那么这时候可能只需要几个专门的 IO 线程和少数来承载用户请求对应的协程的线程，只有当用户有数据传输事件到来的时候才去相应，其他时间直接挂起，这种事件驱动的服务器显然对资源的消耗要小得多
协程库
目前协程库仍然是处于试验阶段，API在版本中不停地被调整，目前可以从这里学习，使用maven、gradle配置，未来会加入到核心库中。
suspend 关键字，用于修饰会被暂停的函数,这些函数只能运行在Continuation或者suspend方法中
第一个协程
fun simpleCorountie() {
launch(CommonPool) {
// 基于线程池创建了一个异步的协程
delay(1000L) //延迟1秒
println("in ${Thread.currentThread().name}") //子线程
println("World!") // 输出
}
println("in ${Thread.currentThread().name}")//主线程
println("Hello") // 主线程中输入hello
Thread.sleep(2000L) //停止2秒
}
fun main(args: Array<String>) {
simpleCorountie()
}
-
我们可以简单的通过
launch方法创建一个协程， -
在效果上很像一个
线程。协程的异步完成基于线程的编程实现的，所以要指定一个CoroutineDispatcher,通常是一个线程池的封装。 -
使用
delay方法进行等待操作，与sleep操作类似，但是仍然不同，因为sleep是就thread而言的，delay专属于corounties
输出如下：
in main
Hello
in ForkJoinPool.commonPool-worker-1
World!
协程的操作
suspend fun simpleCorountie() {
var job= launch(CommonPool) {
// 基于线程池创建了一个异步的协程
delay(1000L) //延迟1秒
println("in ${Thread.currentThread().name}") //子线程
println("World!") // 输出
}
job.join()
println("in ${Thread.currentThread().name}")//主线程
println("Hello") // 主线程中输入hello
Thread.sleep(2000L) //停止2秒
}
fun main(args: Array<String>) {
runBlocking { simpleCorountie() }
}
-
launch方法返回一个
Job对象，通过它可以操作协程 -
通过
join方法，可对协程程进行等待。这块如果了解ForkJoin会比较容易明白 -
也可以通过
cancelAndJoin()方法等待退出 -
通过
cancel方法，可以对线程进行取消,取消后isAlive方法由true变为false -
由于
join()是suspend方法,所以simpleCorountie()也必须是suspend的 -
main方法如果不是suspend,必须通过runBlocking创建一个协程，这个协程使用的是当前线程。
输出如下：
in ForkJoinPool.commonPool-worker-1
World!
in main
Hello
关于runBlocking
fun runBlockTest() {
launch(CommonPool) {
runBlocking {
delay(3000)
println("inner runBlocking： ${Thread.currentThread().name}")
}
delay(1000)
println("should seconds")
println("inner thread： ${Thread.currentThread().name}")
}
println("should first")
runBlocking {
delay(3000)
println("outter runBlocking：in ${Thread.currentThread().name}")
}
println("outter thread： ${Thread.currentThread().name}")
readLine()
}
runBlocking创建一个协程，这个协程使用的是当前线程。
输出：
should first
outter runBlocking：in main
inner runBlocking： ForkJoinPool.commonPool-worker-1
outter thread： main
should seconds
inner thread： ForkJoinPool.commonPool-worker-1
cancelAndJoin()
fun main(args: Array<String>) = runBlocking<Unit> {
val job = launch {
try {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
} finally {
println("I'm running finally") //当finally中无需延迟操作，可以这么写
// withContext(NonCancellable) {//当finally中又延迟操作，需要使用 withContext(NonCancellable)
// println("I'm running finally")
// delay(1000L)
// println("And I've just delayed for 1 sec because I'm non-cancellable")
// }
}
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // 等待finally中的方法执行完毕
// job.cancel () //finally中的方法将没时间执行
println("main: Now I can quit.")
}
-
通常在退出时，
finally可以对资源进行最后的处理和清理，他的运行很重要。 -
使用
cancel()关闭协程，协程内的finally块里的语言可能不执行。 -
使用
cancelAndJoin()可以等待协程执行完后再退出 -
如果
finally需要处理时间，则需要使用withContext(NonCancellable)
withTimeout
fun main(args: Array<String>) = runBlocking<Unit> {
withTimeout(1300L) {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
}
}
- 使用
withTimeout可以创建一个可以超时的代码块
输出：
I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Exception in thread "main" kotlinx.coroutines.experimental.TimeoutCancellationException: Timed out waiting for 1300 MILLISECONDS
- 对于如果存在返回值，可以使用
withTimeoutOrNull,将直接返回null,而不会输出TimeoutCancellationException
fun main(args: Array<String>) = runBlocking<Unit> {
val result = withTimeoutOrNull(1300L) {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
"Done" // 在此之前超时
}
println("Result is $result")
}
输出：
I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Result is null
异步操作
suspend fun doSomethingUsefulOne(): Int {
delay(1000L) // pretend we are doing something useful here
return 13
}
suspend fun doSomethingUsefulTwo(): Int {
delay(1000L) // pretend we are doing something useful here, too
return 29
}
上述代码，我们的方法都是顺序的、同步的操作（按照代码行的顺序执行）
fun main(args: Array<String>) = runBlocking<Unit> {
val time = measureTimeMillis {
val one = doSomethingUsefulOne()//耗时1秒
val two = doSomethingUsefulTwo()//耗时1秒
println("The answer is ${one + two}")//打印时约2秒
}
println("Completed in $time ms")
}
输出:
he answer is 42
Completed in 2017 ms
但是一般独立不相关的两个方法是可以并发的，这个时候我们可以通过async(之前的版本是defer,可能是由于不好理解，就Deprecated了)快速创建异步协程,通过await()方法等待获取数据，直观上可以按照fork and join去理解,它创建了一个Deferred（可以理解为一个轻量的异步返回数据的协程）。
fun main(args: Array<String>) = runBlocking<Unit> {
val time = measureTimeMillis {
val one = async { doSomethingUsefulOne() }//开启一个异步协程获取数据
val two = async { doSomethingUsefulTwo() }//开启另一个异步协程获取数据
println("The answer is ${one.await() + two.await()}")
}
println("Completed in $time ms")
}
输出：
The answer is 42
Completed in 1017 ms
async在默认模式下是立刻执行的，有时候我们会希望在调用await()时再开始启动，获取数据。可以想象一个Thread,在定义过程时、初始化时是不执行的，在start()后才真正执行。这时我们可以使用lazily started async即延迟启动：async(start = CoroutineStart.LAZY)
fun main(args: Array<String>) = runBlocking<Unit> {
val one = async(start = CoroutineStart.LAZY) {
println("corountie started")
"ok"
}
delay(1000)
println("prev call await")
println("The answer is ${one.await()}")
}
输出：
prev call await
corountie started
The answer is ok
CoroutineContext
CoroutineContext是协程的上下文，它由一堆的变量组成，其中包括
- 协程的
Job（launch方法的返回类型，可以被join与cancel），通过corountineContext[Job]可以获取Job对象； CoroutineDispatcher
CoroutineDispatcher
CoroutineDispatcher决定了协程的所在线程，可能指定一个具体的线程，也可能指定一个线程池(CommonPool)，也可能不指定线程，看一个例子：
fun main(args: Array<String>) = runBlocking<Unit> {
val jobs = arrayListOf<Job>()
jobs += launch(Unconfined) { // not confined -- 会在主线程执行
println(" 'Unconfined': I'm working in thread ${Thread.currentThread().name}")
}
jobs += launch(coroutineContext) { // 在当前线程执行
println("'coroutineContext': I'm working in thread ${Thread.currentThread().name}")
}
jobs += launch(CommonPool) { // 将在一个 ForkJoinPool线程池中的线程进行
println(" 'CommonPool': I'm working in thread ${Thread.currentThread().name}")
}
jobs += launch(newSingleThreadContext("MyOwnThread")) { // 创建一个线程
println(" 'newSTC': I'm working in thread ${Thread.currentThread().name}")
}
jobs.forEach { it.join() }
}
输出
'Unconfined': I'm working in thread main
'CommonPool': I'm working in thread ForkJoinPool.commonPool-worker-1
'newSTC': I'm working in thread MyOwnThread
'coroutineContext': I'm working in thread main
-
Unconfined：开始时使用当前线程，但是在使用
delay后可能会转换线程（取决于最后一次调用delay的协程），该模式不适合用在占用CPU时间或者更新UI上 -
coroutineContext：是协程的
CoroutineContext属性，使用当前线程，也是父协程的线程。 -
CommonPool：使用线程池，另外
DefaultDispatcher是用的CommonPool -
newSingleThreadContext("MyOwnThread")：指定一个新线程，这是一个非常耗费资源的方式，要记得在不用时进行关闭。他同时提供了一个
.use方法用于创建协程
fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
fun main(args: Array<String>) {
newSingleThreadContext("Ctx1").use { ctx1 ->
newSingleThreadContext("Ctx2").use { ctx2 ->
runBlocking(ctx1) {
log("Started in ctx1")
withContext(ctx2) {
log("Working in ctx2")
}
log("Back to ctx1")
}
}
}
}
调试协程
在JVM option后添加-Dkotlinx.coroutines.debug,可以开启协程调试信息
fun log(msg: String) = println("[${Thread.currentThread().name}] $msg") //定义一个方法，在`Thread.currentThread().name`将打印出协程的编号
fun main(args: Array<String>) = runBlocking<Unit> {
val a = async(coroutineContext) {
log("I'm computing a piece of the answer")
6
}
val b = async(coroutineContext) {
log("I'm computing another piece of the answer")
7
}
log("The answer is ${a.await() * b.await()}")
}
子协程
-
当协程的
corountineContext用于创建新的协程，新协程将作为原协程的子协程 -
当
父协程被cancel后，子协程会被一起cancel
fun main(args: Array<String>) = runBlocking<Unit> {
// launch a coroutine to process some kind of incoming request
val request = launch {
// it spawns two other jobs, one with its separate context
val job1 = launch {
println("job1: 使用了自己的coroutineContext")
delay(1000)
println("job1: 上层被Cancel了，我还活着...")
}
// and the other inherits the parent context
val job2 = launch(coroutineContext) {
delay(100)
println("job2:我是一个子协程，因为我使用了另一个协程的coroutineContext")
delay(1000)
println("job2:父协程被Cancel了，我也就被cancel了")
}
// request completes when both its sub-jobs complete:
job1.join()
job2.join()
}
delay(500)
request.cancel() // 删除上层协程
delay(1000) //
println("main: 等等看，谁还活着")
}
输出：
job1: 使用了自己的coroutineContext
job2:我是一个子协程，因为我使用了另一个协程的coroutineContext
job1: 上层被Cancel了，我还活着...
main: 等等看，谁还活着
CoroutineContext的 + 操作
如果我们希望在保证父子关系（即父协程被cancel后，子协程也一起被cancel），但是又希望子线程能够在不同的线程运行。我们可以使用+操作符，继承coroutineContext
fun main(args: Array<String>) = runBlocking<Unit> {
val request = launch(coroutineContext) { // use the context of `runBlocking`
val job = launch(coroutineContext + CommonPool) {
println("job: 我是request的子协程因为我使用了他的coroutineContext，但是我的dispatcher是一个CommonPool，yeah~~")
delay(1000)
println("job:父协程被Cancel了，我也就被cancel了，所以你看病毒奥我")
}
job.join() // request completes when its sub-job completes
}
delay(500)
request.cancel() // cancel主协程
delay(1000) //
println("main: 等等看，谁还活着")
}
输出：
job: 我是request的子协程因为我使用了他的coroutineContext，但是我的dispatcher是一个CommonPool，yeah~~
main: 等等看，谁还活着
父协程的join操作对子协程的影响
当父协程执行join操作时，会等待子协程全部执行完毕，才会解除join阻塞，无需在父协程的最后添加子协程的join方法
fun main(args: Array<String>) = runBlocking<Unit> {
val request = launch {
repeat(3) { i ->
// launch a few children jobs
launch(coroutineContext + CommonPool) { //创建了3个子协程，并使用了不一样的dispatcher
delay((i + 1) * 200L) //
println("Coroutine $i is done")
}
}
println("request: 父协程执行完毕")
}
request.join() // 父协程的join会等待所有子协程执行完毕
println("main:子协程终于全部执行完毕了，我可以退出了")
}
输出：
request: 父协程执行完毕
Coroutine 0 is done
Coroutine 1 is done
Coroutine 2 is done
main:子协程终于全部执行完毕了，我可以退出了
协程的基础基本就是这些，后面讲讨论介绍另一个概念Channels
