kotlin最新协程教程——1.2.10

协程与线程的区别:

在高并发的场景下,多个协程可以共享一个或者多个线程,性能可能会要好一些。举个简单的例子,一台服务器有 1k 用户与之连接,如果我们采用类似于 Tomcat 的实现方式,一个用户开一个线程去处理请求,那么我们将要开 1k 个线程,这算是个不小的数目了;而我们如果使用协程,为每一个用户创建一个协程,考虑到同一时刻并不是所有用户都需要数据传输,因此我们并不需要同时处理所有用户的请求,那么这时候可能只需要几个专门的 IO 线程和少数来承载用户请求对应的协程的线程,只有当用户有数据传输事件到来的时候才去相应,其他时间直接挂起,这种事件驱动的服务器显然对资源的消耗要小得多

协程库

目前协程库仍然是处于试验阶段,API在版本中不停地被调整,目前可以从这里学习,使用mavengradle配置,未来会加入到核心库中。

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是协程的上下文,它由一堆的变量组成,其中包括

  • 协程的Joblaunch方法的返回类型,可以被joincancel),通过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

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容