Kotlin协程系列(二)

故乡的樱花开了

发布于 2023-11-28 09:58:38

2530

发布于 2023-11-28 09:58:38

　　在进行业务开发时，我们通常会基于官方的协程框架(kotlinx.coroutines)来运用Kotlin协程优化异步逻辑，不过这个框架过于庞大和复杂，如果直接接触它容易被劝退。所以，为了我们在后续的学习中游刃有余，在使用官方给出的复合协程时能够胸有成竹，我们暂且抛开它，按照它的思路实现一个轻量版的协程框架。

1.开胃小菜：实现一个delay函数

　　在实现delay函数前，我们先来思考delay函数的作用。在使用线程开发时，如果我们想让一段代码延迟一段时间再执行，我们一般会用Thread.sleep函数，但是这个函数的缺点是它会阻塞当前线程。在协程当中，我们同样可以这样做，只是这么做不好，明知道协程可以挂起，却要它阻塞线程，岂不是在浪费cpu资源？

　　我们的目的是让代码延迟一段时间后再执行，只要做到这点就好了。因此，delay函数的实现可以确定以下两点：

不需要阻塞线程
是个挂起函数，指定时间后，能够恢复执行即可

　　这里，直接给出delay函数的实现，然后再作出解释：

suspend fun delay(time:Long,unit: TimeUnit =TimeUnit.MILLISECONDS){
    val executor=Executors.newScheduledThreadPool(1){r->
        Thread(r,"Scheduler").apply{
            isDaemon=true
        }
    }
    if(time<=0){
        return
    }
    suspendCoroutine<Unit> {continuation ->
        executor.schedule({continuation.resume(Unit)},time,unit)
    }
}

　　当time不大于0时，表示无延迟，直接返回就好；接下来需要考虑挂起，我们可以使用suspendCoroutine，不难想到，只要再指定time之后，恢复协程的执行就好，所以只要能够给我们提供一个这样的定时回调机制就可以轻松实现这个功能。

　　在jvm上，我们很自然的就可以想到使用ScheduledExecutorService，问题就这样迎刃而解了。

2.协程的描述

　　客观的讲，startCoroutine和createCoroutine这两个api并不适合直接做业务开发。因此，对于协程的创建，在框架中也要根据不同的目的提供不同的构建器(例如launch,async)，其背后对于封装出来的复合协程的类型描述，就是至关重要的一环。

　　协程的描述类，官方给出的名字是Job，和线程的描述类Thread相比，Job同样有join函数，调用时会挂起协程，直到它的完成，它的cancel函数可以对应Thread的interrupt函数，用于取消协程，isActive可以类比Thread的isActive()，用于查询协程是否还在运行。此外，Job还有取消回调函数invokeOnCancel，完成回调函数invokeOnComplete，用于移除回调的remove函数。

3.协程的创建

　　我们已经给出了协程的描述，知道了协程应该具有哪些能力，接下来就需要如何封装协程的创建了。

　　3.1无返回值的launch函数

　　　　如果一个协程的返回值时Unit，我们可以称它为无返回值的，对于这样的协程，我们只需要启动它即可，下面我们给出launch函数的定义：

fun launch(context: CoroutineContext=EmptyCoroutineContext,block:suspend ()->Unit):Job{
    val completion=StandaloneCoroutine(context)
    block.startCoroutine(completion)
    return completion
}

class StandaloneCoroutine(context:CoroutineContext):AbstractCoroutine<Unit>(context){}

abstract class AbstractCoroutine<T>(context:CoroutineContext):Job,Continuation<T>{
    protected val state=AtomicReference<CoroutineState>()
    override val context:CoroutineContext
    init{
        state.set(CoroutineState.Incomplete())
        this.context=context+this
    }
    val isCompleted
        get()=state.get() is CoroutineState.Complete<*>
    override val isActive:Boolean
        get()=when(state.get()){
            is CoroutineState.Complete<*>->false
            is CoroutineState.Cancelling->false
            else->true
        }
            ......
}

　　　　其中，StandaloneCoroutine是AbstractCoroutine的子类，目前只有一个空实现。

　　3.2实现join函数

　　　　join函数是一个挂起函数，他需要等待协程的执行，此时会有两种情况：被等待的协程已经执行完成，join函数就不会挂起，而是立马返回；被等待的协程尚未完成，此时join将协程挂起，直到协程完成。

　　　　下面给出join函数的伪代码实现：

sealed class CoroutineState {
    class Incomplete():CoroutineState()
    class Cancelling():CoroutineState()
    class Complete():CoroutineState()
}

suspend fun join(){
    when(state.get()){
        is CoroutineState.Incomplete->return joinSuspend()
        is CoroutineState.Cancelling->return joinSuspend()
        is CoroutineState.Complete->return
    }
}

suspend fun joinSuspend()= suspendCoroutine<Unit> {continuation ->   
    doOnCompleted{result->
        continuation.resume(Unit)
    }
}

　　3.3有返回值的async函数

　　　　现在，我们已经知道如何启动协程和等待协程完成了，不过很多时候，我们想拿到协程的返回值，因此我们再基于Job接口再定义一个Deferred接口，如下所示：

interface Deferred<T>:Job{
    suspend fun await()
}

　　　　这里多了一个泛型参数T，T表示返回值类型，通过它的await函数可以拿到这个返回值，因此await函数的主要作用有：在协程执行完成时，立即拿到协程的结果；如果协程尚未完成，则挂起协程，直到它完成，这一点和join类似。下面，给出await函数的定义：

class DeferredCoroutine<T>(context:CoroutineContext):AbstractCoroutine<T>(context),Deferred<T>{
    override suspend fun await():T{
        val currentState=state.get()
        return when(currentState){
            is CoroutineState.Incomplete->awaitSuspend()
            is CoroutineState.Cancelling->awaitSuspend()
            is CoroutineState.Complete->currentState.value as T
        }
    }
    private suspend fun awaitSuspend()=suspendCoroutine<T>{continuation->
        doOnCompleted{result->
            continuation.resumeWith(result)
        }
    }
}

　　　　await函数的实现思路和join函数类似，只是在对结果的处理上有差异。接下来，我们再给出async函数的实现，代码如下：

fun <T> async(context:CoroutineContext=EmptyCoroutineContext,block:suspend ()->T):Deferred<T>{
    val completion=DeferredCoroutine<T>(context)
    block.startCoroutine(completion)
    return completion
}

　　　　这样，我们就可以启动有返回值的协程了，首先定义一个挂起函数，然后用delay(1000)函数来模拟耗时操作，然后我们用async启动协程，并获取协程的返回值，代码如下：

suspend fun getValue():String{
    delay(1000)
    return "使用async启动协程"
}
suspend fun main(){
    val deferred=async{
        getValue()
    }
    val result=deferred.await()
    println(result)//将会打印"使用async启动协程"
}

4.协程的调度

　　截至目前，我们已经大致用程序勾勒出一个比较完善的复合协程了，不过还有一个问题没有解决，我们的协程是如何实现并发的？我们前面在介绍协程的时候提到过协程的挂起和恢复和线程的不同点在于在哪儿挂起，什么时候恢复是开发者自己决定的，这意味着调度工作不能交给操作系统，而应该在用户态解决。

　　协程需要调度的位置就是挂起点的位置，当协程执行到挂起点的位置时，如果产生了异步行为，协程就会在这个挂起点挂起，只有协程在挂起点正真挂起时，我们才有机会实现调度，而实现调度器需要使用协程的拦截器。调度的本质就是解决协程在挂起点恢复后的协程逻辑在哪里运行的问题，由此给出调度器的接口定义：

interface Dispatcher{
  fun dispatch(block:()->Unit)    
}

　　接下来，我们将调度器和拦截器结合起来，拦截器是协程上下文元素的一类实现，下面给出基于调度器的拦截器实现的定义：

open class DispatcherContext(private val dispatcher:Dispatcher):AbstractCoroutineContextElement(ContinuationInterceptor),ContinuationInterceptor{
    override fun <T> interceptorContinuation(continuation:Continuation<T>):Continuation<T>=DispatchedContinuation(continuation,dispatcher)
}
private class DispatchedContinuation<T>(val delegate:Continuation<T>,val dispatcher:Dispatcher):Continuation<T>{
    override val context=delegate.context
    override fun resumeWith(result: Result<T>) {
        dispatcher.dispatch{
            delegate.resumeWith(result)//通过dispatch将协程的恢复执行调度在指定的调度器上
        }
    }
}

　　调度的具体过程其实就是在delegate的恢复调用之前，通过dispatch将其调度在指定的调度器上。下面我们介绍一下有哪些调度器类型：

默认调度器(Dispatchers.Default)：使用共享的线程池，适用于 CPU 密集型的计算任务。默认调度器的线程数量通常与可用的 CPU 核数相等，因此适用于并行计算。但不适合执行可能导致线程阻塞的操作
主线程调度器(Dispatchers.Main)：适用于 Android 应用程序中执行 UI 操作的协程。它会将协程的执行切换到主线程，以确保 UI 操作不会在后台线程上执行
IO调度器(Dispatchers.IO)：专门用于执行涉及阻塞 IO 操作的协程，例如文件读写或网络请求。IO 调度器使用一个专门的线程池，允许执行大量的 IO 操作而不阻塞线程
无限制调度器(Dispatchers.Unconfined)：允许协程在调用挂起函数的线程中继续执行，直到第一个挂起点。之后，它可能在其他线程中继续执行，这取决于具体的挂起函数实现。

5.协程的作用域

　　通常我们提到域，都是用来表示范围的，域既有约束作用，又能提供额外的能力。官方框架在实现复合协程的过程中也提供了作用域，主要用以明确协程之间的父子关系以及对于取消的传播行为。该作用域包括以下三种：

顶级作用域：没有父协程的协程所在的作用域为顶级作用域
协同作用域：协程中启动新的协程，新协程为所在协程的子协程，这种情况下子协程所在的作用域默认为协同作用域。
主同作用域：与协程作用域在协程的父子关系上一致，区别在于处于该作用域下的协程出现未捕获的异常时不会将异常向上传递给父协程。

　　除了这三种作用域中提到的行为外，父子协程之间还存在以下规则：

父协程被取消，则所有的子协程均被取消
父协程需要等待子协程执行完毕后才最终进入完成状态
子协程会继承父协程的协程上下文中的元素，如果自身有相同的key的成员，则覆盖对应的key，覆盖效果仅限自身范围内有效

　　下面给出一个协程作用域的通用接口：

interface CoroutineScope{
  val coroutineContext:CoroutineContext  
}

　　从约束的角度来讲，既然有了作用域，我们就不能任意直接使用launch和async函数来创建协程了，加上作用域之后，我们的launch函数的定义如下：

fun CoroutineScope.launch(context:CoroutineContext=EmptyCoroutineContext,block:suspend CoroutineScope.()->Unit):Job{
    val completion=StandaloneCoroutine(newContext)
    block.startCoroutine(completion,completion)
    return completion
}

　　async函数的实现类似，请自行尝试。

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2023-11-27，如有侵权请联系 cloudcommunity@tencent.com 删除

kotlin