协程的结构化并发：父子关系与取消操作

2023-07-10  本文已影响0人  长点点

前言

在并发编程中，我们经常需要使用多个任务来同时处理不同的工作，以提高程序的效率和响应性。然而，并发编程也带来了很多复杂和困难的问题，比如任务之间的同步和协调，资源和内存的管理，错误和异常的处理等。为了解决这些问题，协程提供了一种结构化并发的概念，让我们可以更容易地编写正确和高效的并发代码。

结构化并发是指协程之间具有父子关系，可以通过父协程控制子协程的状态和行为。这样，我们就不需要手动地管理协程的创建和销毁，也不需要担心协程之间的依赖和冲突。结构化并发可以让我们把注意力集中在业务逻辑上，而不是细节上。

一：父子协程

那么，协程之间是如何建立父子关系的呢？答案是通过Job对象。Job对象是一个表示协程执行状态的对象，它有以下几个属性：

父子协程

当我们创建一个协程时，我们可以指定它的父协程，或者使用当前作用域中的默认父协程。这样，就会在Job对象之间建立父子关系。例如：

// 创建一个顶层作用域
val scope = CoroutineScope(Dispatchers.Default)

// 在作用域中创建一个父协程
val parent = scope.launch {
    // 在父协程中创建一个子协程
    val child = launch {
        // 在子协程中执行一些工作
        println("Child is working...")
        delay(1000)
        println("Child is done.")
    }
    // 等待子协程完成
    child.join()
    println("Parent is done.")
}

在这个例子中，我们在scope作用域中创建了一个父协程parent，并在parent中创建了一个子协程child。这样，parent就会成为child的父协程，并且child会被添加到parent的children属性中。

与线程相比，协程之间的父子关系有很大的不同。线程之间没有父子关系，线程的结束不受其他线程的影响。例如：

// 创建一个线程
Thread parent = new Thread(() -> {
    // 在线程中创建另一个线程
    Thread child = new Thread(() -> {
        // 在子线程中执行一些工作
        System.out.println("Child is working...");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Child is done.");
    });
    // 启动子线程
    child.start();
    // 等待子线程完成
    try {
        child.join();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("Parent is done.");
});
// 启动父线程
parent.start();

在这个例子中，我们创建了一个线程parent，并在parent中创建了另一个线程child。虽然我们可以把parent看作是child的父线程，但实际上它们之间没有任何关系。如果parent被中断或者异常终止，child并不会受到任何影响，它会继续执行直到结束。

多个http请求的请求顺序管理

使用父子协程可以让我们更容易地编写复杂的并发逻辑，比如有依赖关系的多个http请求。例如，假设我们需要发起三个http请求，其中一个请求的参数是另外两个请求的结果。我们可以使用父子协程来实现这个需求，如下：

// 创建一个顶层作用域
val scope = CoroutineScope(Dispatchers.IO)

// 在作用域中创建一个父协程
val parent = scope.launch {
    // 在父协程中创建两个子协程，分别发起第一个和第二个http请求，并返回结果
    val result1 = async { request1() }
    val result2 = async { request2() }
    // 在父协程中等待两个子协程都完成，并获取结果
    val param1 = result1.await()
    val param2 = result2.await()
    // 在父协程中发起第三个http请求，使用前两个请求的结果作为参数，并获取结果
    val result3 = request3(param1, param2)
    // 在父协程中处理最终的结果
    handleResult(result3)
}

在这个例子中，我们在scope作用域中创建了一个父协程parent，并在parent中创建了两个子协程result1和result2，分别发起第一个和第二个http请求，并返回结果。然后，我们在parent中等待两个子协程都完成，并获取它们的结果。接着，我们在parent中发起第三个http请求，使用前两个请求的结果作为参数，并获取结果。最后，我们在parent中处理最终的结果。

使用父子协程的优点有以下几点：

二：父子协程完成操作

有了父子关系，我们就可以利用它来实现一些结构化的并发操作。其中一个操作就是结构化完成操作，也就是说，父协程只有在所有子协程都完成后才算是完成状态。这种操作可以保证协程的执行顺序和结果的一致性。

为了实现结构化完成操作，我们可以使用join方法或者await方法来等待协程的完成。join方法是Job对象的方法，它会挂起当前协程，直到目标协程完成。await方法是Deferred对象的方法，它会挂起当前协程，直到目标协程返回一个结果。Deferred对象是一种特殊的Job对象，它可以表示一个有返回值的协程。

例如：

// 创建一个顶层作用域
val scope = CoroutineScope(Dispatchers.Default)

// 在作用域中创建一个父协程
val parent = scope.launch {
    // 在父协程中创建两个子协程
    val child1 = async {
        // 在子协程中执行一些工作，并返回一个结果
        println("Child1 is working...")
        delay(1000)
        println("Child1 is done.")
        1
    }
    val child2 = launch {
        // 在子协程中执行一些工作
        println("Child2 is working...")
        delay(2000)
        println("Child2 is done.")
    }
    // 等待两个子协程都完成，并获取结果
    val result = child1.await()
    child2.join()
    println("Parent is done. Result is $result.")
}

在这个例子中，我们在scope作用域中创建了一个父协程parent，并在parent中创建了两个子协程child1和child2。child1是一个有返回值的协程，所以我们使用async函数来创建它，并使用await方法来获取它的结果。child2是一个没有返回值的协程，所以我们使用launch函数来创建它，并使用join方法来等待它的完成。这样，parent就会在child1和child2都完成后才结束，并打印出结果。

三：父子协程取消操作

除了结构化完成操作，我们还可以利用父子关系来实现另一个结构化的并发操作，那就是结构化取消操作，也就是说，当父协程取消时，所有子协程也会取消。这种操作可以让我们释放资源和内存，避免内存泄漏和错误。

为了实现结构化取消操作，我们可以使用cancel方法或者cancelAndJoin方法来取消协程。cancel方法是Job对象的方法，它会取消目标协程，并递归地取消所有子协程。cancelAndJoin方法是Job对象的扩展方法，它会先调用cancel方法，然后调用join方法，等待目标协程和所有子协程都完成。

例如：

// 创建一个顶层作用域
val scope = CoroutineScope(Dispatchers.Default)

// 在作用域中创建一个父协程
val parent = scope.launch {
    // 在父协程中创建两个子协程
    val child1 = launch {
        // 在子协程中执行一些工作，并尝试捕获异常
        try {
            println("Child1 is working...")
            delay(1000)
            println("Child1 is done.")
        } catch (e: CancellationException) {
            println("Child1 is cancelled.")
        }
    }
    val child2 = launch {
        // 在子协程中执行一些工作，并尝试捕获异常
        try {
            println("Child2 is working...")
            delay(2000)
            println("Child2 is done.")
        } catch (e: CancellationException) {
            println("Child2 is cancelled.")
        }
    }
    // 取消父协程
    delay(500)
    println("Parent is cancelling...")
    cancelAndJoin()
    println("Parent is cancelled.")
}

在这个例子中，我们在scope作用域中创建了一个父协程parent，并在parent中创建了两个子协程child1和child2。然后，我们在parent中延迟了500毫秒，然后调用cancelAndJoin方法来取消parent和所有子协程。这样，child1和child2都会抛出CancellationException异常，并打印出取消的信息。

结构化取消操作在Android开发中非常有用，可以通过ViewModelScope来管理协程的生命周期，避免内存泄漏。ViewModelScope是一种特殊的作用域，它会在ViewModel被销毁时自动取消所有协程。例如：

class MyViewModel : ViewModel() {
    // 在ViewModelScope中创建一个协程
    fun doSomeWork() = viewModelScope.launch {
        // 在协程中执行一些工作
        println("Work is started...")
        delay(1000)
        println("Work is done.")
    }
}

在这个例子中，我们在MyViewModel类中定义了一个doSomeWork方法，它会在viewModelScope中创建一个协程，并在协程中执行一些工作。当MyViewModel被销毁时，viewModelScope会取消所有协程，释放资源和内存。

四:Job源码分析

Job对象的状态变化

为了更深入地理解Job对象是如何实现协程之间的父子关系和结构化并发操作的，我们可以看一下Job的源码。Job是一个接口，它继承自CoroutineContext.Element，也就是说，它是协程上下文的一个元素。Job接口定义了一些属性和方法，如下：

public interface Job : CoroutineContext.Element {
    // 协程上下文的Key
    public companion object Key : CoroutineContext.Key<Job>

    // 协程是否处于活动状态
    public val isActive: Boolean

    // 协程是否已经完成
    public val isCompleted: Boolean

    // 协程是否已经取消
    public val isCancelled: Boolean

    // 协程的所有子协程
    public val children: Sequence<Job>

    // 取消协程，并传递一个可选的原因
    public fun cancel(cause: CancellationException? = null): Boolean

    // 等待协程完成，并传递一个可选的原因
    public suspend fun join()

    // 注册一个回调函数，在协程完成时调用，并返回一个DisposableHandle对象，可以用来取消注册
    public fun invokeOnCompletion(
        onCancelling: Boolean = false,
        invokeImmediately: Boolean = true,
        handler: CompletionHandler): DisposableHandle

    // 注册一个回调函数，在协程取消时调用，并返回一个DisposableHandle对象，可以用来取消注册
    public fun invokeOnCancellation(handler: CompletionHandler): DisposableHandle

    // 附加一个子协程到当前协程，并返回一个ChildHandle对象，可以用来取消附加
    public fun attachChild(child: Job): ChildHandle

    // 获取协程的取消原因，如果没有则返回null
    public fun getCancellationException(): CancellationException

}

从这些属性和方法中，我们可以看出，Job对象提供了一些基本的功能，比如：

这些功能都是为了实现结构化并发操作所必需的。但是，Job接口只是定义了一些抽象的方法，并没有具体的实现。那么，具体的实现是在哪里呢？

答案是在JobImpl类中。JobImpl类是Job接口的一个实现类，它继承自AbstractCoroutine类，并实现了SelectClause0接口。AbstractCoroutine类是一个抽象类，它封装了一些协程的基本逻辑和状态机制。SelectClause0接口是一个选择器子句接口，它允许我们在select表达式中使用join方法。JobImpl类的代码如下：

internal open class JobImpl(parent: Job? = null) : AbstractCoroutine<Unit>(parent, true), Job, SelectClause0 {
    
    override val onCancelMode: Int get() = ON_CANCEL_MAKE_CANCELLING

    override fun handlesException(): Boolean = true

    override suspend fun join() {
        if (!joinInternal()) { 
            coroutineContext.cancelChildren()
            throw JobCancellationException("Job was cancelled", null, this)
        }
    }

    override fun getCompletionException(): CancellationException =
        state.let { state ->
            when (state) {
                is Finishing -> state.rootCause?.toCancellationException("Job is finishing normally", this)
                is CompletedExceptionally -> state.cause.toCancellationException("Job was cancelled normally", this)
                else -> error("Job is still active or completing: $state")
            }
        }

    override fun invokeOnCompletion(
        onCancelling: Boolean,
        invokeImmediately: Boolean,
        handler: CompletionHandler): DisposableHandle {
        var nodeCache: JobNode<*>? = null 
        while (true) { 
            val state = this.state
            when (state) {
                is Empty -> { 
                    if (state.isActive) { 
                        val node = nodeCache ?: makeNode(handler, onCancelling).also { nodeCache = it }
                        if (_state.compareAndSet(state, node)) return node
                    } else
                        promoteEmptyToNodeList(state) 
                }
                is Incomplete -> { 
                    val list = state.list
                    if (list == null) { 
                        promoteSingleToNodeList(state as JobNode<*>)
                    } else {
                        val node = nodeCache ?: makeNode(handler, onCancelling).also { nodeCache = it }
                        if (addLastAtomic(state, list, node)) return node
                    }
                }
                else -> { 
                    if (invokeImmediately) invokeHandler(handler, state, this)
                    return NonDisposableHandle
                }
            }
        }
    }

    override fun invokeOnCancellation(handler: CompletionHandler): DisposableHandle {
        var nodeCache: JobCancellationNode<*>? = null 
        while (true) { 
            val state = this.state
            when (state) {
                is Empty -> { 
                    if (state.isActive) { 
                        val node = nodeCache ?: JobCancellationNode(this, handler).also { nodeCache = it }
                        if (_state.compareAndSet(state, node)) return node
                    } else
                        promoteEmptyToNodeList(state) 
                }
                is Incomplete -> { 
                    val list = state.list
                    if (list == null) { 
                        promoteSingleToNodeList(state as JobNode<*>)
                    } else {
                        val node = nodeCache ?: JobCancellationNode(this, handler).also { nodeCache = it }
                        if (addLastAtomic(state, list, node)) return node
                    }
                }
                is CompletedExceptionally -> { 
                    handler.invokeIt((state as? CompletedExceptionally)?.cause)
                    return NonDisposableHandle
                }
                else -> return NonDisposableHandle // must be active or cancelled normally
            }
        }
    }

    override fun attachChild(child: Job): ChildHandle {
        var handleCache: ChildHandleNode? = null 
        while (true) {
            val state = this.state
            when (state) {
                is Empty -> { 
                    if (state.isActive) { 
                        val handle =
                            handleCache ?: ChildHandleNode(this, child).also { handleCache = it }
                        if (_state.compareAndSet(state, handle)) return handle
                    } else
                        promoteEmptyToNodeList(state) 
                }
                is Incomplete -> {
                    val list = state.list
                    if (list == null) {
                        promoteSingleToNodeList(state as JobNode<*>)
                    } else {
                        val handle =
                            handleCache ?: ChildHandleNode(this, child).also { handleCache = it }
                        if (addLastAtomic(state, list, handle)) return handle
                    }
                }
                else -> return NonDisposableHandle // must be complete and cannot add child to it anymore
            }
        }
    }

    override fun cancel(cause: CancellationException?) {
        cancelImpl(cause)
    }

    override fun cancelChildren(cause: CancellationException?) {
        this.state.let { state ->
            when {
                state !is Incomplete -> return // fast path when there is no list of children
                cause != null && cause !is CancellationException -> cancelChildrenInternal(state.list, cause)
                else -> cancelChildrenInternal(state.list)
            }
        }
    }

    override fun onCompletionInternal(state: Any?, mode: Int, suppressed: Boolean) {
        if (!suppressed && mode != MODE_IGNORE && state is CompletedExceptionally && handlesException()) {
            cancelParent(cause = state.cause)
        }
    }

    override fun onCancellationInternal(exceptionally: CompletedExceptionally?) {
        cancelChildrenInternal(exceptionally?.cause)
    }

    override fun toString(): String =
        "${nameString()}{${stateString(state)}}@$hexAddress"

    protected open fun nameString(): String =
        "Job"
}

从这些代码中，我们可以看出，JobImpl类主要做了以下几件事：

从上面的分析中，我们可以看出，JobImpl类主要通过以下几种方式来实现协程之间的父子关系和结构化并发操作：

这样，我们就可以利用Job对象来方便地控制协程的状态和行为，实现结构化并发操作。

总结

协程的结构化并发是通过Job对象为我们建立的父子关系，让我们可以方便地控制协程的状态和行为。我们可以使用join方法或者await方法来实现结构化完成操作，保证协程的执行顺序和结果的一致性。我们也可以使用cancel方法或者cancelAndJoin方法来实现结构化取消操作，释放资源和内存，避免内存泄漏和错误。协程的结构化并发可以让我们更容易地编写正确和高效的并发代码，避免错误和内存泄漏。

上一篇下一篇

猜你喜欢

热点阅读