kotlin协程的启动模式 (枚举类CoroutineStart)

CoroutineStart.DEFAULT

*Default——根据协程的上下文立即调度协程执行。* *如果coroutine上下文的[CoroutineDispatcher]从[CoroutineDispatcher]返回' true '。和大多数调度程序一样，
isdispatchrequired] 
*函数的作用是:然后将协调程序代码分派给以后执行，而调用协调程序构建器的代码将继续执行。
* *注意[调度员。unrestricted总是从它的[CoroutineDispatcher]返回' false '。函数，因此使用[Dispatchers启动一个协程。默认情况下unrestricted和使用un分派是一样的。
* *如果coroutine [Job]在有机会开始执行之前就被取消了，那么它根本不会开始它的
*执行，而是会在异常情况下完成。协调程序在挂起点的可取消性取决于
*挂起函数的特定实现细节。使用
[suspend endcancellablecoroutine]实现可取消的挂起函数。

CoroutineStart.LAZY

只有在需要时才懒洋洋地启动协同程序。
*
*有关详细信息，请参阅相应的协程构建器的文档
*(如[发射][CoroutineScope。发射]和[异步][CoroutineScope.async])。
*
*如果coroutine [Job]在有机会开始执行之前就被取消了，那么它将不会开始执行
*完全执行，但会在例外情况下完成。

CoroutineStart.ATOMIC 1.3.31以前试验阶段

自动(即。，以不可取消的方式)根据协程的上下文安排协程的执行。
*这类似于[默认值]，但是协程在开始执行之前不能被取消。
*
*协调程序在暂停点的可取消性取决于特定的实现细节
*暂停函数，如[默认]。

CoroutineStart.UNDISPATCHED 1.3.31以前试验阶段

立即执行协程，直到它的第一个挂起点_in current thread_，就像
*协同程序使用[dispatcher . unrestricted]启动。但是，当协程从暂停状态恢复时
*根据上下文中的[CoroutineDispatcher]进行调度。
*
*这与[ATOMIC]类似，在某种意义上，协程开始执行，即使它已经被取消，
但是不同的是，它开始在同一个线程中执行。
*
*协调程序在暂停点的可取消性取决于特定的实现细节
*暂停函数，如[默认]。
*
** *注意:这是一个实验性的api。在未来使用这种模式时，协程的执行语义可能会发生变化。

job.join() 等待job执行完, 再执行后面的代码
CoroutineStar.DEFAULT模式(如果启动之前cancel, 协程里面的代码执行不到)

fun main() = runBlocking {

    val job: Job = GlobalScope.launch(start = CoroutineStart.DEFAULT) {
        println("1")
        delay(1000L)
        println("2")
    }
//    job.start()
//    job.join()
    println("3")
    delay(2000L)//为了保证结果都能打印, 因为外面的协程1, 不会等待里面的协程2执行完.
}
打印结果, 不加delay(2000L)时:
3
1
打印结果, 加上delay(2000L)时:
3
1
2

CoroutineStart.LAZY模式

fun main() = runBlocking {//协程1

    val job: Job = GlobalScope.launch(start = CoroutineStart.LAZY) {//协程2
        println("1")
        delay(1000L)
        println("2")
    }
    job.start()
    job.join() //等待job执行完, 再执行后面的代码
    println("3")
}
打印结果如下:
1
2
3

Deferred

private suspend fun test2() {
    val deferred1 = GlobalScope.async<Int>(start = CoroutineStart.LAZY) {
        delay(1000L)
        println("计算")
        2
    }
    val deferred2 = GlobalScope.async<Int>(start = CoroutineStart.LAZY) {
        delay(1000L)
        println("计算2")
        3
    }

    val num = deferred1.await() + deferred2.await()//等待deferred1和deferred2两个协程执行完

    println(num)
}

CoroutineStart.DEFAULT和CoroutineStart.ATOMIC区别

suspend fun test3() {
    val job = GlobalScope.launch (start = CoroutineStart.DEFAULT){
        println("1")
        delay(1000L)
        println("2")
    }
    //默认模式 如果执行之前被取消, 里面的内容可能执行不到
    job.cancel()
    println("3")

    val job2 = GlobalScope.launch (start = CoroutineStart.ATOMIC){
        println("4")
        //第一个挂起点
        delay(1000L)
        println("5")
    }
    //自动模式 如果只要执行了, 在第一个挂起点取消才会生效, 也就是4一定会被打印
    job2.cancel()
    println("6")
}

CoroutineStart.UNDISPATCHERED模式

fun main() = runBlocking<Unit> {
    test4()
}

suspend fun test4() {
    //默认和父协程在同一个线程
    val job = GlobalScope.launch(start = CoroutineStart.UNDISPATCHED){
        println("1")
        delay(1000)
        println("2")
    }
    println("5")
    //等待job执行完成, 再执行后面的逻辑
    job.join()
}

协程的调度

协程调度器类型

2019-06-02 at 1.58 PM.png

协程调度器实际上是一个协程拦截器

2019-06-02 at 1.51 PM.png

自定义ContinuationInterceptor和Continuation

fun log(str:String){
    println("${Thread.currentThread().name}, $str")
}
suspend fun main()  {
    val job = GlobalScope.launch(MyContinuationInterceptor() + CoroutineName("HelloWorld")) {
        log("1")
        delay(1000L)
        log("2")
    }

    job.join()
}

class MyContinuationInterceptor : ContinuationInterceptor {
    //ContinuationInterceptor 是ContinuationInterceptor类中的伴生对象
    override val key: CoroutineContext.Key<*> = ContinuationInterceptor
    override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> {
        return MyContinuation(continuation)
    }

}

class MyContinuation<T>(private val continuation: Continuation<T>) : Continuation<T> {

    override val context: CoroutineContext = continuation.context

    private val executor = Executors.newSingleThreadExecutor{ it ->
        Thread(it, "MyThreadExecutor").also { it.isDaemon = true }
    }

    override fun resumeWith(result: Result<T>) {

        executor.submit {
            log("Before...")
            continuation.resumeWith(result)
            log("After...")
        }
    }

}
打印结果:
MyThreadExecutor, Before...
MyThreadExecutor, 1
MyThreadExecutor, After...
MyThreadExecutor, Before...
MyThreadExecutor, 2
MyThreadExecutor, After...

协程的异常处理

协程的异常处理

fun main() = runBlocking<Unit> {
    test11()
    test22()
    test33()
}

private suspend fun test33() {
    GlobalScope.launch {
        val job = launch {
            val num = 3 / 0 //此处发生了异常, 导致父协程被取消, 从而引起job.join()触发已经被取消异常
            delay(1000L)
        }

        try {
            job.join()
        } catch (e: Exception) {//已经被取消异常
            println("test33: $e.localizedMessage")
        }
    }.join()
}

private suspend fun test22() {
    val deferred = GlobalScope.async {//async自己会捕获内部的异常
        val num = 2 / 0
        delay(1000L)
        3
    }

    try {
        val i = deferred.await()
    } catch (e: Exception) {
        println("test22: $e.localizedMessage")
    }
}

private suspend fun test11() {
    val job = GlobalScope.launch(CoroutineExceptionHandler { _, throwable ->
        println("test11: $throwable.localizedMessage")
    }) {//launch 自己不会捕获内部异常
        val num = 3 / 0
        delay(1000L)
    }
    job.join()
}

协程的异常处理2

val demo = Demo1()

fun main() = runBlocking<Unit> {
    demo.test()
}

class Demo1 {
    fun test() = runBlocking {
        GlobalScope.launch(CoroutineExceptionHandler { _, throwable ->
            println("#1 ${throwable.localizedMessage}")
        }) {
            println("1")
            val job = launch(CoroutineExceptionHandler { _, throwable ->
                println("#2 ${throwable.localizedMessage}")
            }) {
                println("2")
                val num = 2 / 0
                println("2 after...")
            }
            println("3")
            job.join()
            println("4")
        }.join()
    }
}

打印结果:
1
3
2
#1 / by zero
异常被父协程捕获, 异常层层向上传递

使用suspervisorScope{}包裹子协程, 子协程自己捕获异常

val demo = Demo1()

fun main() = runBlocking<Unit> {
    demo.test2()
}

class Demo1 {
    fun test2() = runBlocking {
        GlobalScope.launch(CoroutineExceptionHandler { _, throwable ->
            println("#1 ${throwable.localizedMessage}")
        }) {
            println("1")
            supervisorScope {//不想外部传递异常, 但是外部父协程可以取消此\job
                val job = launch(CoroutineExceptionHandler { _, throwable ->
                    println("#2 ${throwable.localizedMessage}")
                }) {
                    println("2")
                    val num = 2 / 0
                    println("2 after...")
                }
            }
            println("3")
        }.join()
    }
}

打印结果:
1
2
#2 / by zero
3

coroutineScope{}继承外部协程上下文环境

val demo = Demo1()

fun main() = runBlocking<Unit> {
    demo.test3()
}
 fun test3() = runBlocking {
        GlobalScope.launch(CoroutineExceptionHandler { _, throwable ->
            println("#1 ${throwable.localizedMessage}")
        }) {
            println("1")
            test4()
            println("3")
        }.join()
    }

    private suspend fun test4(){//继承外部协程环境
        coroutineScope {
            val job = launch(CoroutineExceptionHandler { _, throwable ->
                println("#2 ${throwable.localizedMessage}")
            }) {
                println("2")
                val num = 2 / 0
                println("2 after...")
            }
        }
    }
打印结果:
1
2
#1 / by zero

协程作用域与异常传播

2019-06-02 at 8.24 PM.png

异常捕获处理机制

2019-06-02 at 8.30 PM.png

协程的取消

使用协程封装Retrofit请求数据

    private fun test3() {
        GlobalScope.launch {
            val json = getResult("/adult/user/logon", mutableMapOf("mobile" to "18795455272", "valid_code" to "8158"))
            Log.e("test:: ", json)//打印结果
        }
    }

    private suspend fun getResult(url: String, params: MutableMap<String, String>): String =
            suspendCancellableCoroutine<String> { continuation ->
                val call = RetrofitUtils.getApiService().post(url, params)
                //失败取消请求
                continuation.invokeOnCancellation { call.cancel() }
                call.enqueue(object : retrofit2.Callback<JsonObject> {
                    override fun onFailure(call: Call<JsonObject>, t: Throwable) {
                        continuation.resumeWithException(t)
                    }

                    override fun onResponse(call: Call<JsonObject>, response: Response<JsonObject>) {

                        if (response.isSuccessful) {
                            response.body()?.toString()?.let { continuation.resume(it) }
                                    ?: continuation.resumeWithException(NullPointerException("json is null"))

                        } else {
                            continuation.resumeWith(Result.failure(HttpException(response)))
                        }
                    }

                })
            }