Kotlin-通过async与await实现高效并发

2021-06-15 本文已影响0人蒋斌文

Kotlin-通过async与await实现高效并发

image-20210615150418309

/*
    挂起函数的组合
 */

fun main() = runBlocking {
    val elapsedTime = measureTimeMillis {
        val value1 = intValue1()
        val value2 = intValue2()

        println("$value1 + $value2 = ${value1 + value2}")
    }

    println("total time: $elapsedTime")
}



private suspend fun intValue1(): Int {
    delay(1000)
    return 15
}

private suspend fun intValue2(): Int {
    delay(2000)
    return 20
}

RUN> 🏂🏂🏂🏂🏂🏂
15 + 20 = 35
total time: 3010

Process finished with exit code 0
总耗时3秒多(2+1),说明程序是串行执行的

这俩函数之间是没有任何依赖关系的，那如果有一种机制能在执行intValue1的同时，也能并行的执行intValue2,那最终总的执行效率是不是会大幅度提升呢?

使用 async 并发

image-20210615150802387

/*
    使用async与await实现并发

    从概念上来说，async就像是launch一样。它会开启一个单独的协程，这个协程是个轻量级线程，可以与其他协程并发工作。区别在于，launch
    会返回一个Job，但是Job并不会持有任何结果值，而async会返回一个Deferred，这是一个轻量级的非阻塞的future，它代表一个promise，可以
    在稍后提供一个结果值。

    可以通过在一个deferred值上调用.await()方法来获取最终的结果值，Deferred也是个Job，因此可以在需要时对其进行取消
 */

fun main() = runBlocking {
    val elapsedTime = measureTimeMillis {
        val value1 = async { intValue1() }
        val value2 = async { intValue2() }


        val result1 = value1.await()
        val result2 = value2.await()

        println("$result1 + $result2 = ${result1 + result2}")
    }

    println("total time: $elapsedTime")

}

private suspend fun intValue1(): Int {
    delay(2000)
    return 15
}

private suspend fun intValue2(): Int {
    delay(1000)
    return 20
}

RUN> 🏄🏄🏄🏄🏄🏄
15 + 20 = 35
total time: 2021

Process finished with exit code 0
总耗时2秒多,说明程序是并行执行的
image-20210615151457111

public fun <T> CoroutineScope.async(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> T
): Deferred<T> {
    val newContext = newCoroutineContext(context)
    val coroutine = if (start.isLazy)
        LazyDeferredCoroutine(newContext, block) else
        DeferredCoroutine<T>(newContext, active = true)
    coroutine.start(start, coroutine, block)
    return coroutine
}

image-20210615151137938

Deferred就是一个Job

image-20210615151233515

惰性启动的 async

image-20210615151610474

async是接收三个参数的，另外两个参数有默认值：

public fun <T> CoroutineScope.async(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> T
): Deferred<T> {
    val newContext = newCoroutineContext(context)
    val coroutine = if (start.isLazy)
        LazyDeferredCoroutine(newContext, block) else
        DeferredCoroutine<T>(newContext, active = true)
    coroutine.start(start, coroutine, block)
    return coroutine
}

image-20210615151755225

Default -- immediately schedules the coroutine for execution according to its context.

根据上下文会立即调用协程执行

也就是上一次咱们做的这个async中的代码块是立即会执行的，不是延时执行的，但是！！在实际场景中可以会有遇到async时不希望立刻就来执行，而是这个执行的时机由程序员在某个条件下来触发，此时这个CoroutineStart参数就需要改变一下不能用默认参数了，需要用它了：LAZY

/**
     * Starts the coroutine lazily, only when it is needed.
     *
     * See the documentation for the corresponding coroutine builders for details
     * (like [launch][CoroutineScope.launch] and [async][CoroutineScope.async]).
     *
     * If coroutine [Job] is cancelled before it even had a chance to start executing, then it will not start its
     * execution at all, but will complete with an exception.
     */
LAZY,

接下来就是要来启动延迟的async进行执行了

/*
    关于async的延迟执行

    我们可以通过将async方法的start参数设置为CoroutineStart.LAZY来实现协程的延迟执行。
    在这种情况下，协程会在两种场景下去执行：调用Deferred的await方法，或是调用Job的start方法。

 */
fun main() = runBlocking {
    val elapsedTime = measureTimeMillis {
        val value1 = async(start = CoroutineStart.LAZY) { intValue1() }
        val value2 = async(start = CoroutineStart.LAZY) { intValue2() }

        println("hello world")
//        Thread.sleep(3000)
        delay(3000L)
        // 尝试注释掉如下两行代码
        value1.start()
        value2.start()
        val result1 = value1.await()
        val result2 = value2.await()

        println("$result1 + $result2 = ${result1 + result2}")
    }
    println("total time: $elapsedTime")
}


private suspend fun intValue1(): Int {
    delay(2000)
    return 15
}

private suspend fun intValue2(): Int {
    delay(1000)
    return 20
}

RUN> 🏂🏂🏂🏂🏂🏂
hello world
15 + 20 = 35
total time: 5046
delay(3000L)+intValue1(2000) = 5秒，程序是并行执行的

image-20210615152334555

fun main() = runBlocking {
    val elapsedTime = measureTimeMillis {
        val value1 = async(start = CoroutineStart.LAZY) { intValue1() }
        val value2 = async(start = CoroutineStart.LAZY) { intValue2() }

        println("hello world")
//        Thread.sleep(3000)
        delay(3000L)
        // 尝试注释掉如下两行代码
//        value1.start()
//        value2.start()
        val result1 = value1.await()
        val result2 = value2.await()

        println("$result1 + $result2 = ${result1 + result2}")
    }
    println("total time: $elapsedTime")
}


private suspend fun intValue1(): Int {
    delay(2000)
    return 15
}

private suspend fun intValue2(): Int {
    delay(1000)
    return 20
}

RUN> 🏄🏄🏄🏄🏄🏄
hello world
15 + 20 = 35
total time: 6027

Process finished with exit code 0
delay(3000L)+intValue1(2000) +intValue2(1000) 程序串行执行

其实根源是：
image-20210615152635899

所以，如果将async变成了lazy了之后需要特别的注意！！！如果使用不当的时候跟不用async没任何区别了，这样就失去了async的使用意义了。

async ⻛格的函数

image-20210615152955681

/*
    异步风格的函数
 */


fun main() {
    val elapsedTime = measureTimeMillis {
        val value1 = intValue1Async()
        val value2 = intValue2Async()

        runBlocking {
            println("the answer is: ${value1.await() + value2.await()}")
        }
    }

    println("total time: $elapsedTime")
}


private suspend fun intValue1(): Int {
    delay(2000)
    return 15
}

private suspend fun intValue2(): Int {
    delay(3000)
    return 20
}

fun intValue1Async() = GlobalScope.async {
    intValue1()
}

fun intValue2Async() = GlobalScope.async {
    intValue2()
}

RUN> 🏄🏄🏄🏄🏄🏄
the answer is: 35
total time: 3161

Process finished with exit code 0
image-20210615153228945

结构化并发程序开发：

/*
    使用async进行结构化并发程序开发
 */

fun main() = runBlocking {
    val elapsedTime = measureTimeMillis {
        println("the answer is: ${intSum()}")
    }

    println("total time: $elapsedTime")
}


private suspend fun intSum(): Int = coroutineScope {
    val value1 = async { intValue1() }
    val value2 = async { intValue2() }

    value1.await() + value2.await()
}


private suspend fun intValue1(): Int {
    delay(2000)
    return 15
}
private suspend fun intValue2(): Int {
    delay(3000)
    return 20
}

RUN> 🏂🏂🏂🏂🏂🏂

the answer is: 35
total time: 3019

Process finished with exit code 0

取消始终通过协程的层次结构来进行传递:

image-20210615153625963

/*
    关于父子协程的异常与取消问题

    协程的取消总是会沿着协程层次体系向上进行传播
 */

fun main() = runBlocking<Unit> {
    try {
        failureComputation()
    } finally {
        println("Computation failed")
    }
}


private suspend fun failureComputation(): Int = coroutineScope {
    val value1 = async<Int> {
        try {
            delay(9000000)
            50
        } finally {
            println("value1 was cancelled")
        }
    }

    val value2 = async<Int> {
        Thread.sleep(2000)
        println("value2 throws an exception")

        throw Exception()
    }
    value1.await() + value2.await()
}

RUN> 🏂🏂🏂🏂🏂🏂

value2 throws an exception
value1 was cancelled
Computation failed
Exception in thread "main" java.lang.Exception
  at com.shengsiyuan.coroutines3.HelloKotlin6Kt$failureComputation$2$value2$1.invokeSuspend(HelloKotlin6.kt:37)
  at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
  at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
  at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:274)
  at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:84)
  at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:59)
  at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
  at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:38)
  at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
  at com.shengsiyuan.coroutines3.HelloKotlin6Kt.main(HelloKotlin6.kt:14)
  at com.shengsiyuan.coroutines3.HelloKotlin6Kt.main(HelloKotlin6.kt)

Process finished with exit code 1

image-20210615154029585

“协程的取消总是会沿着协程层次体系向上进行传播。”，以上就是解决在async中如果出异常的一个比较好的解决方案。

Kotlin-通过async与await实现高效并发

Kotlin-通过async与await实现高效并发

使用 async 并发

惰性启动的 async

async ⻛格的函数

结构化并发程序开发：

取消始终通过协程的层次结构来进行传递:

猜你喜欢

热点阅读