Kotlin协程(2)
Kotlin协程(2) 异常处理的必要性 @OptIn(ExperimentalStdlibApi::class) // 标记该代码使用了实验性标准库 API @Test // 表示这是一个测试函数 fun `test CoroutineContext extend2`() = runBlocking { // 使用 runBlocking 启动协程 // 创建一个协程异常处理器，用于捕获协程中的异常 val coroutineExceptionHandler = CoroutineExceptionHandler { _, exception -> println("Caught : $exception") // 打印捕获到的异常 } // 创建一个协程作用域，包含一个 Job，使用主线程调度器和异常处理器 val scope = CoroutineScope( Job() + Dispatchers.Main + coroutineExceptionHandler // Job() 用于管理协程的生命周期 ) // 在 IO 线程上启动一个新的协程 val job = scope.launch(Dispatchers.IO) { // 获取当前协程上下文中的调度器 val dispatcher = coroutineContext[CoroutineDispatcher] // 打印当前使用的调度器 println("Current dispatcher: $dispatcher") } // 等待协程完成 job.join() // 确保主协程在子协程完成后再继续 } 异常的传播 协程构建器有两种形式：1、自动传播异常（launch与actor）2、向用户暴露异常（async与produce）， 当这些构建器用于创建一个根协程时（该协程不是另一个协程的子协程），前者这类构建器，异常会在它发生的第一时间被抛出，而后者则依赖用户来最终消费异常，例如通过await或receive。 // 根协程 @Test @OptIn(DelicateCoroutinesApi::class) fun `test exception propagation`() = runBlocking<Unit> { // 在 GlobalScope 中启动一个新协程，返回一个 Job 对象 val job = GlobalScope.launch { try { // 在协程中抛出一个 IndexOutOfBoundsException throw IndexOutOfBoundsException() } catch (e: Exception) { // 捕获异常并打印一条消息，指示捕获了 IndexOutOfBoundsException println("IndexOutOfBoundsException") // 如果需要可以选择打印堆栈轨迹 // e.printStackTrace() } } // 等待协程完成 job.join() // 在 GlobalScope 中启动一个新协程，并返回一个 Deferred 对象 val deffer = GlobalScope.async { // 在协程中抛出一个 ArithmeticException throw ArithmeticException() } try { // 等待异步任务完成并获取结果，如果任务抛出异常，这里会重新抛出异常 deffer.await() } catch (e: Exception) { // 捕获异常并打印一条消息，指示捕获了 ArithmeticException println("ArithmeticException") // 如果需要可以选择打印堆栈轨迹 // e.printStackTrace() } } 非根协程的异常 其他协程所创建的协程中，产生的异常总是会被传播 /* 非根协程，job是scope内的子协程，所以会直接消费掉异常 */ @Test fun `test exception propagation2`() = runBlocking { // 创建一个新的 CoroutineScope，使用 Job 作为其上下文元素 val scope = CoroutineScope(Job()) // 在 scope 中启动一个新协程，返回一个 Job 对象 val job = scope.launch { // 在协程中启动一个异步任务 async { // 在异步任务中抛出一个 IllegalArgumentException throw IllegalArgumentException() } } // 等待 launch 启动的协程完成 job.join() } 异常的传播特性 当一个协程由于一个异常而运行失败时，它会传播这个异常并传递给它的父级。接下来，父级会进行下面几步操作： 取消它自己的子级 取消它自己 将异常传播并传递给它的父级 SupervisorJob 使用SupervisorJob时，一个子协程的运行失败不会影响其他子协程，SupervisorJob不会传播异常给它的父级，它会让子协程自己处理异常。 需求：在作用域内定义作业的UI组件，如果任何一个UI的子作业执行失败了，它并不总是有必要取消整个UI组件，但是如果UI组件被销毁了，由于它的结果不再被需要了，它就有必要使所有的子作业执行失败。 @Test fun `test supervisorJob`() = runBlocking<Unit> { // 创建一个新的 CoroutineScope，使用 SupervisorJob 作为其上下文元素 val supervisor = CoroutineScope(SupervisorJob()) // 如果使用普通的 Job，注释掉上一行并启用以下行 // val supervisor = CoroutineScope(Job()) // 在 supervisor 作用域中启动第一个子协程，返回一个 Job 对象 val job1 = supervisor.launch { // 延迟 100 毫秒以模拟一些工作 delay(100) println("child 1") // 在第一个子协程中抛出一个 IllegalArgumentException throw IllegalArgumentException() } // 在 supervisor 作用域中启动第二个子协程，返回一个 Job 对象 val job2 = supervisor.launch { try { // 无限延迟以模拟长时间运行的任务 delay(Long.MAX_VALUE) } finally { // 当协程取消时，最终会打印 "child 2 finished" println("child 2 finished") } } // 以下是可选部分，如果需要，可以取消注释 // 延迟 200 毫秒后取消 supervisor 作用域内的所有协程 // delay(200) // supervisor.cancel() // 等待 job1 和 job2 都完成 joinAll(job1, job2) } SupervisorScope 当作业自身执行失败的时候，所有子作业将会被全部取消 @Test fun `test supervisorScope`() = runBlocking<Unit> { // 使用 supervisorScope 启动一个新的协程作用域 supervisorScope { // 在 supervisorScope 中启动一个子协程，返回一个 Job 对象 launch { // 延迟 100 毫秒以模拟一些工作 delay(100) // 输出 "child 1"，表示子协程已完成其任务 println("child 1") // 在协程中抛出一个 IllegalArgumentException，以测试异常传播 throw IllegalArgumentException() } // 在 supervisorScope 中启动另一个协程 try { // 延迟一个非常长的时间，模拟长时间运行的任务 delay(Long.MAX_VALUE) } finally { // 当 supervisorScope 被取消时（由于异常），最终会打印 "child 2 finished." println("child 2 finished.") } } } @Test fun `test supervisorScope2`() = runBlocking<Unit> { // 使用 supervisorScope 来启动一个协程作用域 supervisorScope { // 在 supervisorScope 中启动一个子协程 val child = launch { try { // 输出 "The child is sleeping" 表示子协程进入睡眠状态 println("The child is sleeping") // 延迟无限长的时间，模拟长时间运行的任务 delay(Long.MAX_VALUE) } finally { // 当子协程被取消时，最终会打印 "The child is cancelled" println("The child is cancelled") } } // 让出线程以确保子协程启动并进入延迟状态 yield() // 在 supervisorScope 中抛出一个异常 println("Throwing an exception from the scope") throw AssertionError() // 抛出一个断言错误，用于测试作用域的异常处理 } } 异常的捕获 使用CoroutineExceptionHandler对协程的异常进行捕获 以下的条件被满足时，异常就会被捕获： 时机：异常是被自动抛出异常的协程所抛出的（使用launch，而不是async时）； 位置：在CoroutineScope的CoroutineContext中或在一个根协程（CoroutineScope或者supervisorScope的直接子协程）中。 Android中全局异常处理 全局异常处理器可以获取到所有协程未处理的未捕获异常，不过它并不能对异常进行捕获，虽然不能阻止程序崩溃，全局异常处理器在程序调试和异常上报等场景中仍然有非常大的用处。 我们需要在classpath下面创建META-INF/services目录，并在其中创建一个名为kotlinx.coroutines.CoroutineExceptionHandler的文件，文件内容就是我们的全局异常处理器的全类名。 取消与异常 取消与异常紧密相关，协程内部使用CancellationException来进行取消，这个异常会被忽略。 当子协程被取消时，不会取消它的父协程 如果一个协程遇到了CancellationException以外的异常，它将使用该异常取消它的父协程。当父协程的所有子协程都结束后，异常才会被父协程处理。 @Test fun `test cancel and exception`() = runBlocking<Unit> { // 启动一个协程作为父协程 val job = launch { // 在父协程中启动一个子协程 val child = launch { try { // 子协程尝试延迟执行，模拟长时间运行的任务 delay(Long.MAX_VALUE) } finally { // 当子协程被取消时，执行此块 println("Child is cancelled") } } // 让父协程暂时让出线程，给其他协程执行的机会 yield() println("Cancelling child") // 取消子协程，并等待其执行完 finally 块 child.cancelAndJoin() // 再次让出线程 yield() println("Parent is not cancelled") } // 等待父协程完成执行 job.join() } @Test fun `test cancel and exception2`() = runBlocking<Unit> { // 定义一个 CoroutineExceptionHandler 来处理未捕获的异常 val handler = CoroutineExceptionHandler { _, exception -> println("Caught: $exception") } // 使用 GlobalScope 启动一个顶层协程，并应用异常处理器 val job = GlobalScope.launch(handler) { // 第一个子协程，模拟一个需要很长时间的任务 launch { try { // 使用一个很长的延迟来模拟长时间运行的任务 delay(Long.MAX_VALUE) } finally { // 确保在协程被取消时，执行以下代码 withContext(NonCancellable) { println("Children are cancelled, but exceptions are not handled") delay(100) // 确保输出语句有机会执行 println("The first child finished its non cancellable block") } } } // 第二个子协程，在短暂延迟后抛出一个异常 launch { delay(10) println("Second child throws an exception") throw ArithmeticException() // 抛出一个异常来测试异常处理 } } // 等待父协程完成 job.join() } 异常聚合 当协程的多个子协程因为异常而失败时，一般情况下取第一个异常进行处理。在第一异常之后发生的所有其他异常，都将绑定到第一个异常之上。 @Test fun `test exception aggregation`() = runBlocking<Unit> { // 定义一个 CoroutineExceptionHandler 来处理未捕获的异常 val handler = CoroutineExceptionHandler { _, exception -> // 输出被捕获的主异常和抑制的异常（如果有） println("Caught: $exception, Suppressed: ${exception.suppressed.contentToString()}") } // 使用 GlobalScope 启动一个顶层协程，并应用异常处理器 val job = GlobalScope.launch(handler) { // 第一个子协程 launch { try { // 模拟长时间运行的任务 delay(Long.MAX_VALUE) } finally { // 在取消时抛出 ArithmeticException throw ArithmeticException() } } // 第二个子协程 launch { try { // 进行短暂的延迟 delay(100) } finally { // 在取消时抛出 IOException throw IOException() } } // 第三个子协程 launch { try { // 进行更短暂的延迟 delay(10) } finally { // 在取消时抛出 IndexOutOfBoundsException throw IndexOutOfBoundsException() } } } // 等待父协程完成，确保所有子协程的异常都被处理 job.join() } Flow-异步流 // 定义一个名为 simpleFlow 的挂起函数，返回一个 Flow<Int> suspend fun simpleFlow() = flow<Int> { // 使用 for 循环遍历从 1 到 100 的整数 for (i in 1..100) { delay(1000) // 延迟 1 秒钟，模拟异步任务的延迟 emit(i) // 向下游发射当前整数 } } // 这是一个测试函数，演示了异步、非阻塞式的行为 @Test fun `test multiple values3`() = runBlocking<Unit> { // 启动一个新的协程，不阻塞主线程 launch { // 使用 for 循环打印信息，展示主线程没有被阻塞 for (k in 1..3) { println("I am not blocked $k") delay(1500) // 延迟 1.5 秒钟 } } // 收集 simpleFlow() 发射的值 simpleFlow().collect { value -> println(value) // 打印每个收集到的值 } } 冷流 Flow是一种类似于序列的冷流，flow构建器中的代码直到流被收集的时候才运行 // 测试函数，演示 Flow 是冷流的概念 @Test fun `test flow is cold`() = runBlocking { // 创建一个新的 Flow 实例 val flow = simpleFlow2() // 首次调用 collect，开始收集 Flow 发射的值 println("Calling collect...") flow.collect { value -> println(value) // 打印每个收集到的值 } // 再次调用 collect，重新收集 Flow 发射的值 println("Calling collect again...") flow.collect { value -> println(value) // 打印每个收集到的值 } } 流的连续性 流的每次单独收集都是按顺序执行的，除非使用特殊操作符 从上游到下游每个过渡操作符都会处理每个发射出的值，然后再交给末端操作符。 // 测试函数，演示流的连续操作 @Test fun `test flow continuation`() = runBlocking { // 将 1 到 5 的整数转换成流 (1..5).asFlow() // 过滤流，只保留偶数 .filter { it % 2 == 0 } // 将每个整数转换为字符串 .map { "String $it" } // 收集流中的每个元素 .collect { value -> println("Collect $value") // 打印收集到的每个元素 } } 流构建器 flowOf构建器定义了一个发射固定值集的流。 @Test fun `test flow continuation`() = runBlocking { // 将 1 到 5 的整数转换成流 (1..5).asFlow() // 过滤流，只保留偶数 .filter { it % 2 == 0 } // 将每个整数转换为字符串 .map { "String $it" } // 收集流中的每个元素 .collect { value -> println("Collect $value") // 打印收集到的每个元素 } } 使用.asFlow()扩展函数，可以将各种集合与序列转换为流。 @Test fun `test flow builder2`() = runBlocking<Unit> { // 创建一个包含三个字符串元素的 Flow："one"、"two" 和 "three" flowOf("one", "two", "three") // 对 Flow 中的每个元素，应用 1000 毫秒（1 秒）的延迟 .onEach { delay(1000) } // 收集 Flow 中的每个元素，并将其打印到控制台 .collect { v -> println(v) } // 将整数范围从 1 到 3 转换为 Flow 并收集每个元素 (1..3).asFlow() .collect { println(it) // 将每个收集到的整数打印到控制台 } } 流的上下文 流的收集总是在调用协程的上下文中发生，流的该属性称为上下文保存 flow{...}构建器中的代码必须遵循上下文保存属性，并且不允许从其他上下文中发射（emit） flowOn操作符，该函数用于更改流发射的上下文。 // 定义一个简单的 Flow，返回 Int 类型的值 fun simpleFlow5() = flow<Int> { // 打印当前线程的名称，表明 Flow 开始执行 println("Flow started ${Thread.currentThread().name}") // 生成从 1 到 3 的整数 for (i in 1..3) { // 延迟 1000 毫秒（1 秒）模拟耗时操作 delay(1000) // 发射（emit）当前整数 emit(i) } // 设置 Flow 在默认的调度器（Dispatcher）上运行 }.flowOn(Dispatchers.Default) // 测试函数，使用 JUnit 测试框架 @Test fun `test flow context3`() = runBlocking<Unit> { // 收集 Flow 中发射的值 simpleFlow5().collect { v -> // 打印当前线程的名称和收集到的值 println("Collected $v ${Thread.currentThread().name}") } } 启动流 使用launchIn替换collect，可以在单独的协程中启动流的收集 ...