kotlin最新協(xié)程教程——1.2.10

協(xié)程與線程的區(qū)別:

在高并發(fā)的場景下,多個(gè)協(xié)程可以共享一個(gè)或者多個(gè)線程,性能可能會(huì)要好一些。舉個(gè)簡單的例子,一臺(tái)服務(wù)器有 1k 用戶與之連接,如果我們采用類似于 Tomcat 的實(shí)現(xiàn)方式,一個(gè)用戶開一個(gè)線程去處理請(qǐng)求,那么我們將要開 1k 個(gè)線程,這算是個(gè)不小的數(shù)目了;而我們?nèi)绻褂脜f(xié)程,為每一個(gè)用戶創(chuàng)建一個(gè)協(xié)程,考慮到同一時(shí)刻并不是所有用戶都需要數(shù)據(jù)傳輸,因此我們并不需要同時(shí)處理所有用戶的請(qǐng)求,那么這時(shí)候可能只需要幾個(gè)專門的 IO 線程和少數(shù)來承載用戶請(qǐng)求對(duì)應(yīng)的協(xié)程的線程,只有當(dāng)用戶有數(shù)據(jù)傳輸事件到來的時(shí)候才去相應(yīng),其他時(shí)間直接掛起,這種事件驅(qū)動(dòng)的服務(wù)器顯然對(duì)資源的消耗要小得多

協(xié)程庫

目前協(xié)程庫仍然是處于試驗(yàn)階段,API在版本中不停地被調(diào)整,目前可以從這里學(xué)習(xí),使用maven、gradle配置,未來會(huì)加入到核心庫中。

suspend 關(guān)鍵字,用于修飾會(huì)被暫停的函數(shù),這些函數(shù)只能運(yùn)行在Continuation或者suspend方法中

第一個(gè)協(xié)程

fun simpleCorountie() {

    launch(CommonPool) {
        // 基于線程池創(chuàng)建了一個(gè)異步的協(xié)程
        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方法創(chuàng)建一個(gè)協(xié)程,

  • 在效果上很像一個(gè)線程。協(xié)程的異步完成基于線程的編程實(shí)現(xiàn)的,所以要指定一個(gè)CoroutineDispatcher,通常是一個(gè)線程池的封裝。

  • 使用delay方法進(jìn)行等待操作,與sleep操作類似,但是仍然不同,因?yàn)?code>sleep是就thread而言的,delay專屬于corounties

輸出如下:

in main
Hello
in ForkJoinPool.commonPool-worker-1
World!

協(xié)程的操作

suspend fun simpleCorountie() {

 var job=  launch(CommonPool) {
        // 基于線程池創(chuàng)建了一個(gè)異步的協(xié)程
        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方法返回一個(gè)Job對(duì)象,通過它可以操作協(xié)程

  • 通過join方法,可對(duì)協(xié)程程進(jìn)行等待。這塊如果了解ForkJoin會(huì)比較容易明白

  • 也可以通過cancelAndJoin()方法等待退出

  • 通過cancel方法,可以對(duì)線程進(jìn)行取消,取消后isAlive方法由true變?yōu)?code>false

  • 由于join()suspend方法,所以simpleCorountie() 也必須是suspend

  • main方法如果不是suspend,必須通過runBlocking創(chuàng)建一個(gè)協(xié)程,這個(gè)協(xié)程使用的是當(dāng)前線程。

輸出如下:

in ForkJoinPool.commonPool-worker-1
World!
in main
Hello

關(guān)于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創(chuàng)建一個(gè)協(xié)程,這個(gè)協(xié)程使用的是當(dāng)前線程。

輸出:

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") //當(dāng)finally中無需延遲操作,可以這么寫
       // withContext(NonCancellable) {//當(dāng)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中的方法執(zhí)行完畢
   // job.cancel ()  //finally中的方法將沒時(shí)間執(zhí)行
    println("main: Now I can quit.")
}
  • 通常在退出時(shí),finally可以對(duì)資源進(jìn)行最后的處理和清理,他的運(yùn)行很重要。

  • 使用cancel()關(guān)閉協(xié)程,協(xié)程內(nèi)的finally塊里的語言可能不執(zhí)行。

  • 使用cancelAndJoin()可以等待協(xié)程執(zhí)行完后再退出

  • 如果finally需要處理時(shí)間,則需要使用withContext(NonCancellable)

withTimeout

fun main(args: Array<String>) = runBlocking<Unit> {
    withTimeout(1300L) {
        repeat(1000) { i ->
            println("I'm sleeping $i ...")
            delay(500L)
        }
    }
}
  • 使用withTimeout 可以創(chuàng)建一個(gè)可以超時(shí)的代碼塊

輸出:

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
  • 對(duì)于如果存在返回值,可以使用withTimeoutOrNull,將直接返回null,而不會(huì)輸出TimeoutCancellationException
fun main(args: Array<String>) = runBlocking<Unit> {
    val result = withTimeoutOrNull(1300L) {
        repeat(1000) { i ->
            println("I'm sleeping $i ...")
            delay(500L)
        }
        "Done" // 在此之前超時(shí)
    }
    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
}

上述代碼,我們的方法都是順序的、同步的操作(按照代碼行的順序執(zhí)行)

fun main(args: Array<String>) = runBlocking<Unit> {
    val time = measureTimeMillis {
        val one = doSomethingUsefulOne()//耗時(shí)1秒
        val two = doSomethingUsefulTwo()//耗時(shí)1秒
        println("The answer is ${one + two}")//打印時(shí)約2秒
    }
    println("Completed in $time ms")
}

輸出:

he answer is 42
Completed in 2017 ms

但是一般獨(dú)立不相關(guān)的兩個(gè)方法是可以并發(fā)的,這個(gè)時(shí)候我們可以通過async(之前的版本是defer,可能是由于不好理解,就Deprecated了)快速創(chuàng)建異步協(xié)程,通過await()方法等待獲取數(shù)據(jù),直觀上可以按照fork and join去理解,它創(chuàng)建了一個(gè)Deferred(可以理解為一個(gè)輕量的異步返回?cái)?shù)據(jù)的協(xié)程)。

fun main(args: Array<String>) = runBlocking<Unit> {
   val time = measureTimeMillis {
       val one = async { doSomethingUsefulOne() }//開啟一個(gè)異步協(xié)程獲取數(shù)據(jù)
       val two = async { doSomethingUsefulTwo() }//開啟另一個(gè)異步協(xié)程獲取數(shù)據(jù)
       println("The answer is ${one.await() + two.await()}")
   }
   println("Completed in $time ms")
}

輸出:

The answer is 42
Completed in 1017 ms

async在默認(rèn)模式下是立刻執(zhí)行的,有時(shí)候我們會(huì)希望在調(diào)用await()時(shí)再開始啟動(dòng),獲取數(shù)據(jù)??梢韵胂笠粋€(gè)Thread,在定義過程時(shí)、初始化時(shí)是不執(zhí)行的,在start()后才真正執(zhí)行。這時(shí)我們可以使用lazily started async延遲啟動(dòng)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是協(xié)程的上下文,它由一堆的變量組成,其中包括

  • 協(xié)程的Joblaunch方法的返回類型,可以被joincancel),通過corountineContext[Job]可以獲取Job對(duì)象;
  • CoroutineDispatcher

CoroutineDispatcher

CoroutineDispatcher決定了協(xié)程的所在線程,可能指定一個(gè)具體的線程,也可能指定一個(gè)線程池(CommonPool),也可能不指定線程,看一個(gè)例子:


fun main(args: Array<String>) = runBlocking<Unit> {
    val jobs = arrayListOf<Job>()
    jobs += launch(Unconfined) { // not confined -- 會(huì)在主線程執(zhí)行
        println("      'Unconfined': I'm working in thread ${Thread.currentThread().name}")
    }
    jobs += launch(coroutineContext) { // 在當(dāng)前線程執(zhí)行
        println("'coroutineContext': I'm working in thread ${Thread.currentThread().name}")
    }
    jobs += launch(CommonPool) { // 將在一個(gè) ForkJoinPool線程池中的線程進(jìn)行
        println("      'CommonPool': I'm working in thread ${Thread.currentThread().name}")
    }
    jobs += launch(newSingleThreadContext("MyOwnThread")) { // 創(chuàng)建一個(gè)線程
        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:開始時(shí)使用當(dāng)前線程,但是在使用delay后可能會(huì)轉(zhuǎn)換線程(取決于最后一次調(diào)用delay的協(xié)程),該模式不適合用在占用CPU時(shí)間或者更新UI

  • coroutineContext:是協(xié)程的CoroutineContext屬性,使用當(dāng)前線程,也是父協(xié)程的線程。

  • CommonPool:使用線程池,另外DefaultDispatcher是用的CommonPool

  • newSingleThreadContext("MyOwnThread"):指定一個(gè)新線程,這是一個(gè)非常耗費(fèi)資源的方式,要記得在不用時(shí)進(jìn)行關(guān)閉。他同時(shí)提供了一個(gè).use方法用于創(chuàng)建協(xié)程

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")
            }
        }
    }
}

調(diào)試協(xié)程

JVM option后添加-Dkotlinx.coroutines.debug,可以開啟協(xié)程調(diào)試信息

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg") //定義一個(gè)方法,在`Thread.currentThread().name`將打印出協(xié)程的編號(hào)

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()}")
}

子協(xié)程

  • 當(dāng)協(xié)程的corountineContext用于創(chuàng)建新的協(xié)程,新協(xié)程將作為原協(xié)程子協(xié)程

  • 當(dāng)父協(xié)程cancel后, 子協(xié)程會(huì)被一起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:我是一個(gè)子協(xié)程,因?yàn)槲沂褂昧肆硪粋€(gè)協(xié)程的coroutineContext")
            delay(1000)
            println("job2:父協(xié)程被Cancel了,我也就被cancel了")
        }
        // request completes when both its sub-jobs complete:
        job1.join()
        job2.join()
    }
    delay(500)
    request.cancel() // 刪除上層協(xié)程
    delay(1000) // 
    println("main: 等等看,誰還活著")
}

輸出:

job1: 使用了自己的coroutineContext
job2:我是一個(gè)子協(xié)程,因?yàn)槲沂褂昧肆硪粋€(gè)協(xié)程的coroutineContext
job1: 上層被Cancel了,我還活著...
main: 等等看,誰還活著

CoroutineContext的 + 操作

如果我們希望在保證父子關(guān)系(即父協(xié)程cancel后,子協(xié)程也一起被cancel),但是又希望子線程能夠在不同的線程運(yùn)行。我們可以使用+操作符,繼承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的子協(xié)程因?yàn)槲沂褂昧怂腸oroutineContext,但是我的dispatcher是一個(gè)CommonPool,yeah~~")
            delay(1000)
            println("job:父協(xié)程被Cancel了,我也就被cancel了,所以你看病毒奧我")
        }
        job.join() // request completes when its sub-job completes
    }
    delay(500)
    request.cancel() // cancel主協(xié)程
    delay(1000) //
    println("main: 等等看,誰還活著")
}

輸出:

job: 我是request的子協(xié)程因?yàn)槲沂褂昧怂腸oroutineContext,但是我的dispatcher是一個(gè)CommonPool,yeah~~
main: 等等看,誰還活著

父協(xié)程的join操作對(duì)子協(xié)程的影響

當(dāng)父協(xié)程執(zhí)行join操作時(shí),會(huì)等待子協(xié)程全部執(zhí)行完畢,才會(huì)解除join阻塞,無需在父協(xié)程的最后添加子協(xié)程join方法

fun main(args: Array<String>) = runBlocking<Unit> {
    val request = launch {
        repeat(3) { i ->
            // launch a few children jobs
            launch(coroutineContext + CommonPool) { //創(chuàng)建了3個(gè)子協(xié)程,并使用了不一樣的dispatcher
                delay((i + 1) * 200L) // 
                println("Coroutine $i is done")
            }
        }
        println("request: 父協(xié)程執(zhí)行完畢")
    }
    request.join() // 父協(xié)程的join會(huì)等待所有子協(xié)程執(zhí)行完畢
    println("main:子協(xié)程終于全部執(zhí)行完畢了,我可以退出了")
}

輸出:

request: 父協(xié)程執(zhí)行完畢
Coroutine 0 is done
Coroutine 1 is done
Coroutine 2 is done
main:子協(xié)程終于全部執(zhí)行完畢了,我可以退出了

協(xié)程的基礎(chǔ)基本就是這些,后面講討論介紹另一個(gè)概念Channels

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點(diǎn),簡書系信息發(fā)布平臺(tái),僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容