Kotlin語言基礎(chǔ)筆記
Kotlin流程控制語句筆記
Kotlin操作符重載與中綴表示法筆記
Kotlin擴(kuò)展函數(shù)和擴(kuò)展屬性筆記
Kotlin空指針安全(null-safety)筆記
Kotlin類型系統(tǒng)筆記
Kotlin面向?qū)ο缶幊坦P記
Kotlin委托(Delegation)筆記
Kotlin泛型型筆記
Kotlin函數(shù)式編程筆記
Kotlin與Java互操作筆記
Kotlin協(xié)程筆記
很多小伙伴可能會(huì)覺得Java有了線程丛晦、線程池了,我們還要協(xié)程(Coroutines)干嘛。這里還是有些區(qū)別的腹纳。區(qū)別有:
- 線程是為了提高CPU的利用率雷滚,調(diào)度是由操作系統(tǒng)決定的酒朵,而協(xié)程是為了解決多個(gè)任務(wù)更好的協(xié)作初斑,調(diào)度是由我們代碼控制的宪肖。
- 協(xié)程并不是為了取代線程疯搅,協(xié)程對(duì)線程進(jìn)行抽象濒生,你可以看成協(xié)程是一個(gè)異步調(diào)用的框架,解決了之前線程間協(xié)作代碼繁瑣的問題幔欧。
我們先來看一段代碼罪治,如下:
data class Product(var id: String, var title: String)
data class Stock(var pid: String, var stock: Int)
data class Pms(var pid: String, var pmsTips: String)
suspend fun getProductsByIds(pids: List<String>): List<Product> {
delay(1000)
return listOf(Product("1", "a"), Product("2", "b"))
}
suspend fun getProductStocksByIds(pids: List<String>): List<Stock> {
delay(2000)
return listOf(Stock("1", 2), Stock("2", 4))
}
suspend fun getProductPMSByIds(pids: List<String>): List<Pms> {
delay(3000)
return listOf(Pms("1", "100減99"), Pms("2", "100減99"))
}
fun combine(products: List<Product>?, productStocks: List<Stock>?, productPMS: List<Pms>?) {
println(products)
println(productStocks)
println(productPMS)
}
fun main(args: Array<String>) = runBlocking<Unit> {
val pids = listOf<String>("1", "2")
val products = async {
withTimeoutOrNull(1500) {
getProductsByIds(pids)
}
}
val productStocks = async {
withTimeoutOrNull(2500) {
getProductStocksByIds(pids)
}
}
val productPMS = async {
withTimeoutOrNull(2500) {
getProductPMSByIds(pids)
}
}
val measureTimeMillis = measureTimeMillis {
combine(products.await(), productStocks.await(), productPMS.await())
}
println(measureTimeMillis)
}
這段代碼看起來就像是偽代碼,不過還是非常容易理解礁蔗,就是通過一批商品id觉义,分別調(diào)用三個(gè)接口拿到商品的信息,商品的庫存浴井,商品的優(yōu)惠信息晒骇,然后再合并數(shù)據(jù),這個(gè)場(chǎng)景無論在后端還是前端都會(huì)經(jīng)常遇到磺浙,比如APP調(diào)用的一個(gè)接口洪囤,需要從不同的底層系統(tǒng)獲取到不同部分的數(shù)據(jù),然后聚合好一次性返回給APP屠缭。想想如果是用Java來實(shí)現(xiàn)的會(huì)有多復(fù)雜箍鼓。用Kotlin的協(xié)程實(shí)現(xiàn)就像是寫順序執(zhí)行的代碼,但實(shí)際上你做的是異步調(diào)用呵曹。
1.第一個(gè)協(xié)程代碼
fun main(args: Array<String>) {
launch { // launch new coroutine in background and continue
delay(1000L) // non-blocking delay for 1 second (default time unit is ms)
println("World!") // print after delay
}
println("Hello,") // main thread continues while coroutine is delayed
Thread.sleep(2000L) // block main thread for 2 seconds to keep JVM alive
}
我們使用lauch
來啟動(dòng)一個(gè)協(xié)程款咖,其中要注意的是delay
這個(gè)函數(shù),看起來它跟Thread.sleep
是一樣的作用奄喂,但是他們有本質(zhì)的區(qū)別铐殃,Thread.sleep
會(huì)阻塞當(dāng)前線程(線程就傻傻的在等待),而delay
是暫停當(dāng)前的協(xié)程跨新,不會(huì)阻塞當(dāng)前線程富腊,這個(gè)線程可以去做其他事情。delay
是一個(gè)suspending function域帐,它只能運(yùn)行在協(xié)程里面赘被,如果不在協(xié)程中運(yùn)行是整,會(huì)報(bào)以下異常。
Error: Kotlin: Suspend functions are only allowed to be called from a coroutine or another suspend function
2. runBlocking
runBlocking
函數(shù)會(huì)阻塞當(dāng)前線程民假,一直等到協(xié)程運(yùn)行完浮入。上面的例子可以改成下面的:
fun main(args: Array<String>) = runBlocking<Unit> { // start main coroutine
launch { // launch new coroutine in background and continue
delay(1000L)
println("World!")
}
println("Hello,") // main coroutine continues here immediately
delay(2000L) // delaying for 2 seconds to keep JVM alive
}
3.等待協(xié)程完成
延時(shí)一段時(shí)間來等待協(xié)程完成通常不是很高效,我們可以通過join
來實(shí)現(xiàn)一旦協(xié)程完成就退出main函數(shù)羊异。
fun main(args: Array<String>) = runBlocking<Unit> {
val job = launch { // launch new coroutine and keep a reference to its Job
delay(1000L)
println("World!")
}
println("Hello,")
job.join() // wait until child coroutine completes
}
4. suspending function 暫停函數(shù)
我們也可以使用suspending function重構(gòu)下事秀。
fun main(args: Array<String>) = runBlocking<Unit> {
val job = launch { doWorld() }
println("Hello,")
job.join()
}
// this is your first suspending function
suspend fun doWorld() {
delay(1000L)
println("World!")
}
注意:delay
也是一個(gè)suspending function,所以depay
只能放在suspending function或者協(xié)程代碼(lanuch)里面野舶。
5. 協(xié)程是非常輕量級(jí)的
fun main(args: Array<String>) = runBlocking<Unit> {
val jobs = List(100_000) { // launch a lot of coroutines and list their jobs
launch {
delay(1000L)
print(".")
}
}
jobs.forEach { it.join() } // wait for all jobs to complete
}
啟動(dòng)了10萬個(gè)協(xié)程易迹,最后代碼能夠成功的執(zhí)行完成。同樣平道,大家可以試試換成起10萬個(gè)線程試試睹欲,應(yīng)該會(huì)得出OOM的結(jié)果。
6. 協(xié)程像守護(hù)線程
請(qǐng)看下面這段代碼:
fun main(args: Array<String>) = runBlocking<Unit> {
launch {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
}
delay(1300L) // just quit after delay
}
輸出如下:
I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
可以知道巢掺,等待1.3秒后句伶,main退出了。不會(huì)等待launch的協(xié)程運(yùn)行完陆淀。
7. 協(xié)程取消
launch
返回一個(gè)Job對(duì)象考余,它可以被取消:
fun main(args: Array<String>) = runBlocking<Unit> {
val job = launch {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancel() // cancels the job
job.join() // waits for job's completion
println("main: Now I can quit.")
}
輸出如下:
I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
main: I'm tired of waiting!
main: Now I can quit.
可以看到,一旦調(diào)用了job.cancel()轧苫,就退出了main函數(shù)楚堤。Job還有一個(gè)cancelAndJoin方法,合并了cancel和join操作含懊。
8. 協(xié)程的取消可能需要協(xié)作完成
協(xié)程的取消可能需要協(xié)作完成身冬,所有在kotlinx.coroutines
包下面的suspending functions都可以被取消,但是如果一個(gè)協(xié)程處在計(jì)算中岔乔,他是不能被取消的酥筝,比如這個(gè)例子:
fun main(args: Array<String>) = runBlocking<Unit> {
val startTime = System.currentTimeMillis()
val job = launch {
var nextPrintTime = startTime
var i = 0
while (i < 5) { // computation loop, just wastes CPU
// print a message twice a second
if (System.currentTimeMillis() >= nextPrintTime) {
println("I'm sleeping ${i++} ...")
nextPrintTime += 500L
}
}
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")
}
你可以看到調(diào)用取消后,還在打印雏门。
9. 讓處于計(jì)算中的協(xié)程可取消
有兩種方式可以做到:
- 最簡(jiǎn)單的在while循環(huán)最后面調(diào)用下yield函數(shù)嘿歌。這樣就在每次循環(huán)后讓協(xié)程有了被取消的機(jī)會(huì)茁影,yield是
kotlinx.coroutines
包下的suspending functions宙帝。 - 檢查協(xié)程取消的狀態(tài),如果發(fā)現(xiàn)被取消募闲,則退出循環(huán)步脓。
下面我們以第二種方式演示下:
fun main(args: Array<String>) = runBlocking<Unit> {
val startTime = System.currentTimeMillis()
val job = launch {
var nextPrintTime = startTime
var i = 0
while (isActive) { // cancellable computation loop
// print a message twice a second
if (System.currentTimeMillis() >= nextPrintTime) {
println("I'm sleeping ${i++} ...")
nextPrintTime += 500L
}
}
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")
isActive是協(xié)程的CoroutineScope的一個(gè)屬性。
10. 協(xié)程中try catch finally
當(dāng)協(xié)程被取消時(shí),catch和finally可以被執(zhí)行靴患。
fun main(args: Array<String>) = runBlocking<Unit> {
val job = launch {
try {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
}catch (e:Throwable){
println("I'm running catch")
} finally {
println("I'm running finally")
}
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")
}
輸出:
I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
main: I'm tired of waiting!
I'm running catch
I'm running finally
main: Now I can quit.
11. withContext函數(shù)
在上個(gè)例子中仍侥,如果我們?cè)趂inally塊中調(diào)用suspending functions的話,會(huì)拋出CancellationException蚁廓,因?yàn)閰f(xié)程已經(jīng)被取消了访圃。不過一般來說沒什么太大問題,只要不調(diào)用suspending functions相嵌。如果你一定要在調(diào)用的話,你可以使用withContext(NonCancellable) {...}
况脆。如下:
fun main(args: Array<String>) = runBlocking<Unit> {
val job = launch {
try {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
} 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() // cancels the job and waits for its completion
println("main: Now I can quit.")
}
12. Timeout超時(shí)
如果要設(shè)定協(xié)程調(diào)用超時(shí)時(shí)間饭宾,我們可以使用withTimeout函數(shù),如下:
fun main(args: Array<String>) = runBlocking<Unit> {
withTimeout(1300L) {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
}
}
輸出如下:
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
如果超時(shí)的時(shí)候你不想拋出異常格了,你可以使用withTimeoutOrNull函數(shù)看铆,超時(shí)的時(shí)候它會(huì)返回null。
fun main(args: Array<String>) = runBlocking<Unit> {
val result = withTimeoutOrNull(1300L) {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
"Done" // will get cancelled before it produces this result
}
println("Result is $result")
}
輸出如下:
I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Result is null
13. 使用async并發(fā)調(diào)用
async與launch類似盛末,它也是啟動(dòng)一個(gè)協(xié)程弹惦,只不過lauch返回的是Job(沒有返回值),而async返回的是Deferred(帶返回值)悄但,你可以使用.await()
來獲取Deferred的值棠隐。Deferred是Job的子類,所以Deferred也可以被取消檐嚣≈螅看看下面這段代碼:
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 = 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 1016 ms
因?yàn)槭遣⑿姓{(diào)用,所以時(shí)間差不多是1秒嚎京。
14. async延時(shí)調(diào)用
fun main(args: Array<String>) = runBlocking<Unit> {
val time = measureTimeMillis {
val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
println("The answer is ${one.await() + two.await()}")
}
println("Completed in $time ms")
}
如果async帶上了start = CoroutineStart.LAZY
參數(shù)嗡贺,協(xié)程不會(huì)立即執(zhí)行,會(huì)等到調(diào)用await的時(shí)候才開始執(zhí)行鞍帝。上面代碼輸出如下:
The answer is 42
Completed in 2017 ms
執(zhí)行結(jié)果看起來變成了順序執(zhí)行诫睬,那是因?yàn)閛ne.await執(zhí)行完成之后,才會(huì)開始調(diào)用two.await()執(zhí)行帕涌。所以變成了順序執(zhí)行摄凡。
15. Async-style functions
// The result type of somethingUsefulOneAsync is Deferred<Int>
fun somethingUsefulOneAsync() = async {
doSomethingUsefulOne()
}
// The result type of somethingUsefulTwoAsync is Deferred<Int>
fun somethingUsefulTwoAsync() = async {
doSomethingUsefulTwo()
}
上面兩個(gè)方法xxxAsync
并不是suspending functions,所以他們可以在任何地方調(diào)用宵膨。
// note, that we don't have `runBlocking` to the right of `main` in this example
fun main(args: Array<String>) {
val time = measureTimeMillis {
// we can initiate async actions outside of a coroutine
val one = somethingUsefulOneAsync()
val two = somethingUsefulTwoAsync()
// but waiting for a result must involve either suspending or blocking.
// here we use `runBlocking { ... }` to block the main thread while waiting for the result
runBlocking {
println("The answer is ${one.await() + two.await()}")
}
}
println("Completed in $time ms")
}
16. Dispatchers and threads
launch和async都接收一個(gè)可選的CoroutineContext參數(shù)可以用來指定CoroutineDispatcher架谎。如下:
fun main(args: Array<String>) = runBlocking<Unit> {
val jobs = arrayListOf<Job>()
jobs += launch(Unconfined) { // not confined -- will work with main thread
println(" 'Unconfined': I'm working in thread ${Thread.currentThread().name}")
}
jobs += launch(coroutineContext) { // context of the parent, runBlocking coroutine
println("'coroutineContext': I'm working in thread ${Thread.currentThread().name}")
}
jobs += launch(CommonPool) { // will get dispatched to ForkJoinPool.commonPool (or equivalent)
println(" 'CommonPool': I'm working in thread ${Thread.currentThread().name}")
}
jobs += launch(newSingleThreadContext("MyOwnThread")) { // will get its own new thread
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
默認(rèn)的dispatcher是DefaultDispatcher當(dāng)前的實(shí)現(xiàn)是CommonPool
17. Unconfined vs confined dispatcher
Unconfined dispatcher會(huì)在當(dāng)前線程開始執(zhí)行協(xié)程,但是僅僅是在第一個(gè)暫停點(diǎn)辟躏,之后它恢復(fù)后的dispatcher取決于那個(gè)線程執(zhí)行suspending function谷扣。
coroutineContext 是CoroutineScope的一個(gè)屬性,它的dispatcher會(huì)繼承它parent線程的dispatcher。 代碼如下:
fun main(args: Array<String>) = runBlocking<Unit> {
val jobs = arrayListOf<Job>()
jobs += launch(Unconfined) { // not confined -- will work with main thread
println(" 'Unconfined': I'm working in thread ${Thread.currentThread().name}")
delay(500)
println(" 'Unconfined': After delay in thread ${Thread.currentThread().name}")
}
jobs += launch(coroutineContext) { // context of the parent, runBlocking coroutine
println("'coroutineContext': I'm working in thread ${Thread.currentThread().name}")
delay(1000)
println("'coroutineContext': After delay in thread ${Thread.currentThread().name}")
}
jobs.forEach { it.join() }
}
輸出如下:
'Unconfined': I'm working in thread main
'coroutineContext': I'm working in thread main
'Unconfined': After delay in thread kotlinx.coroutines.DefaultExecutor
'coroutineContext': After delay in thread main
所以会涎,coroutineContext繼承了runBlocking的main線程裹匙,而unconfined恢復(fù)后變成了default executor線程。
18. 線程切換
加上-Dkotlinx.coroutines.debug
JVM參數(shù)運(yùn)行下面的代碼:
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")
}
}
}
}
這里展示了幾個(gè)用法:一個(gè)是使用runBlocking指明一個(gè)特殊的Context末秃,另外一個(gè)是使用withContext來切換Context概页,輸出如下:
[Ctx1 @coroutine#1] Started in ctx1
[Ctx2 @coroutine#1] Working in ctx2
[Ctx1 @coroutine#1] Back to ctx1
還有就是run
來釋放線程。
19. 通過Context來獲取Job
協(xié)程的Job是Context的一個(gè)屬性练慕,如下:
fun main(args: Array<String>) = runBlocking<Unit> {
println("My job is ${coroutineContext[Job]}")
}
19. 子協(xié)程
在協(xié)程中使用coroutineContext來啟動(dòng)另一個(gè)協(xié)程惰匙,新協(xié)程的Job變成了父協(xié)程的子Job,當(dāng)父協(xié)程取消時(shí)铃将,子協(xié)程也會(huì)被取消项鬼。
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: I have my own context and execute independently!")
delay(1000)
println("job1: I am not affected by cancellation of the request")
}
// and the other inherits the parent context
val job2 = launch(coroutineContext) {
delay(100)
println("job2: I am a child of the request coroutine")
delay(1000)
println("job2: I will not execute this line if my parent request is cancelled")
}
// request completes when both its sub-jobs complete:
job1.join()
job2.join()
}
delay(500)
request.cancel() // cancel processing of the request
delay(1000) // delay a second to see what happens
println("main: Who has survived request cancellation?")
}
輸出結(jié)果如下:
job1: I have my own context and execute independently!
job2: I am a child of the request coroutine
job1: I am not affected by cancellation of the request
main: Who has survived request cancellation?
20. Context聯(lián)合
協(xié)程Context可以使用+
聯(lián)合,如下:
fun main(args: Array<String>) = runBlocking<Unit> {
// start a coroutine to process some kind of incoming request
val request = launch(coroutineContext) { // use the context of `runBlocking`
// spawns CPU-intensive child job in CommonPool !!!
val job = launch(coroutineContext + CommonPool) {
println("job: I am a child of the request coroutine, but with a different dispatcher")
delay(1000)
println("job: I will not execute this line if my parent request is cancelled")
}
job.join() // request completes when its sub-job completes
}
delay(500)
request.cancel() // cancel processing of the request
delay(1000) // delay a second to see what happens
println("main: Who has survived request cancellation?")
}
job是request的子協(xié)程劲阎,但是是在CommonPool的線程中執(zhí)行操作绘盟。所以取消request,job也會(huì)取消悯仙。
21. 父協(xié)程會(huì)等待子協(xié)程完成
父協(xié)程會(huì)等待子協(xié)程完成龄毡,不需要使用join來等待他們完成。
fun main(args: Array<String>) = runBlocking<Unit> {
// launch a coroutine to process some kind of incoming request
val request = launch {
repeat(3) { i -> // launch a few children jobs
launch(coroutineContext) {
delay((i + 1) * 200L) // variable delay 200ms, 400ms, 600ms
println("Coroutine $i is done")
}
}
println("request: I'm done and I don't explicitly join my children that are still active")
}
request.join() // wait for completion of the request, including all its children
println("Now processing of the request is complete")
}
輸出如下:
request: I'm done and I don't explicitly join my children that are still active
Coroutine 0 is done
Coroutine 1 is done
Coroutine 2 is done
Now processing of the request is complete
22. Tricks
假如我們現(xiàn)在在寫一個(gè)anroid app锡垄,在activity中啟動(dòng)了很多協(xié)程異步調(diào)用接口獲取數(shù)據(jù)沦零,當(dāng)這個(gè)activity被destory后,所有的協(xié)程需要被取消偎捎,要不然就可能會(huì)發(fā)生內(nèi)存泄漏蠢终。
我們可以創(chuàng)建一個(gè)Job實(shí)例,然后使用launch(coroutineContext, parent = job)
來明確指定parent job茴她。
這樣的話寻拂,我們可以調(diào)用Job.cancel來取消所有的子協(xié)程,而Job.join可以等待所有的子協(xié)程完成丈牢。如下:
fun main(args: Array<String>) = runBlocking<Unit> {
val job = Job() // create a job object to manage our lifecycle
// now launch ten coroutines for a demo, each working for a different time
val coroutines = List(10) { i ->
// they are all children of our job object
launch(coroutineContext, parent = job) { // we use the context of main runBlocking thread, but with our parent job
delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
println("Coroutine $i is done")
}
}
println("Launched ${coroutines.size} coroutines")
delay(500L) // delay for half a second
println("Cancelling the job!")
job.cancelAndJoin() // cancel all our coroutines and wait for all of them to complete
}
輸出如下:
Launched 10 coroutines
Coroutine 0 is done
Coroutine 1 is done
Cancelling the job!
23. channel, select, actor
請(qǐng)看:https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md