# Kotlin 协程 ## 简介 Kotlin 协程为异步编程提供了一个强大的框架，该框架是轻量级的、富有表现力的，并且基于结构化并发原则构建。协程允许编写外观和行为类似于顺序代码的异步代码，消除回调地狱并提高可读性。与线程不同，协程非常轻量级——数百万个协程可以在有限的资源上运行。协程框架包括用于非阻塞操作的挂起函数、用于启动工作的构建器、用于反应流的 Flow 以及全面的取消和异常处理机制。该技能涵盖 Android 开发和服务器端 Kotlin 的协程基础知识、构建器、上下文、流程、通道和生产模式。 ## 暂停函数挂起函数是协程的构建块，可实现可以暂停和恢复而不阻塞线程的非阻塞操作。 // 基本挂起函数暂停乐趣 fetchUser(id: Int): User { 延迟（1000）//挂起而不阻塞返回用户（id，“爱丽丝”） } 数据类用户（val id：Int，val name：String） // 调用挂起函数挂起乐趣 loadUserProfile(id: Int): UserProfile { val 用户 = fetchUser(id) val posts = fetchPosts(user.id) 返回用户配置文件（用户，帖子） } 暂停乐趣 fetchPosts(userId: Int): List { 延迟(500) 返回空列表() } 数据类 Post(val id: Int, val title: String) 数据类 UserProfile(val user: User, val posts: List) // 顺序执行 vs 并发执行暂停乐趣 loadDataSequential(): Pair> { val 用户 = fetchUser(1) val posts = fetchPosts(1) 将用户返回到帖子 } 挂起乐趣 loadDataConcurrent(): Pair> = coroutineScope { val userDeferred = 异步 { fetchUser(1) } val postsDeferred = async { fetchPosts(1) } userDeferred.await() 到 postsDeferred.await() } // 通过回调暂停函数挂起乐趣 fetchData(url: String): String = suspendCoroutine { 继续 -> fetchDataWithCallback(url) { 结果，错误 -> 如果（错误！=空）{ 继续.resumeWithException（错误） } 否则{ 继续.resume(结果) } } } 有趣的 fetchDataWithCallback( 网址：字符串，回调：（字符串，异常？）-> 单位）{ // 模拟基于回调的API 回调（“数据”，空） } // 可取消的挂起函数暂停有趣的 downloadFile(url: String): ByteArray = suspendCancellableCoroutine { 继续 -> val request = startDownload(url) { 数据, 错误 -> 如果（错误！=空）{ 继续.resumeWithException（错误） } 否则{ 继续.resume(数据) } } 延续.invokeOnCancellation { request.cancel() } } 类下载请求{ 有趣的取消() {} } fun startDownload(url: String, 回调: (ByteArray, Exception?) -> Unit): 下载请求 { 返回下载请求() } // 使用withContext进行调度程序切换暂停乐趣 saveToDatabase(用户: User) { withContext(Dispatchers.IO) { // IO调度程序上的数据库操作 println("正在保存用户：${user.name}") } } 暂停有趣的updateUI（用户：用户）{ withContext(Dispatchers.Main) { // 主线程更新UI println("正在更新用户界面：${user.name}") } } 挂起函数标有“挂起”修饰符，并且只能从其他挂起函数或协程中调用，以确保正确的上下文。 ## 协程构建器协程构建器启动具有不同生命周期和结果处理语义的协程，从而实现结构化和非结构化并发。 // 启动：即发即忘协程有趣的启动示例（）{ GlobalScope.launch { val 用户 = fetchUser(1) println("用户：${用户名}") } } // 异步：带有结果的协程有趣的 asyncExample() { GlobalScope.launch { val deferredUser = 异步 { fetchUser(1) } val deferredPosts = async { fetchPosts(1) } val 用户 = deferredUser.await() val posts = deferredPosts.await() println("已加载${user.name}的${posts.size}个帖子") } } // runBlocking：桥接阻塞和暂停世界有趣的 runBlockingExample() = runBlocking { val 用户 = fetchUser(1) println("用户已加载：${user.name}") } // coroutineScope：结构化并发挂起乐趣 loadMultipleUsers(ids: List): List = coroutineScope { ids.map { id -> 异步 { fetchUser(id) } }.awaitAll() } // SupervisorScope：独立子失败暂停乐趣 loadDataWithSupervisor(): List = SupervisorScope { val user1 = 异步 { fetchUser(1) } val user2 = 异步 { 延迟(100) 抛出异常（“失败”） } // 即使 user2 失败，user1 也会成功非空列表（尝试 { user1.await() } catch (e: 异常) { null } ） } // withTimeout：限时协程暂停乐趣 fetchWithTimeout(id: Int)：用户？ { 返回尝试{ 超时（2000）{ 获取用户（id） } } catch (e: TimeoutCancellationException) { 空 } } // 具有生命周期的结构化并发类 ViewModel : CoroutineScope { 私有 val 作业 = SupervisorJob() 覆盖 val coroutineContext: CoroutineContext get() = Dispatchers.Main + 作业有趣的负载数据（）{ 发射{ val 用户 = fetchUser(1) // 更新用户界面 } } 有趣的 onCleared() { 作业.取消() } } 与“coroutineScope”的结构化并发可确保子协程在作用域退出之前完成，从而防止泄漏并确保正确的清理。 ## 协程上下文和调度程序协程上下文定义了执行环境，包括调度程序、作业、异常处理程序和用于调试的协程名称。 // 线程池的调度程序暂停有趣的dispatcherExamples() { // 主要：UI 线程 (Android/JavaFX) withContext(Dispatchers.Main) { println("在主线程上：${Thread.currentThread().name}") } // IO：用于阻塞I/O操作 withContext(Dispatchers.IO) { println("在 IO 线程上：${Thread.currentThread().name}") } // 默认：CPU 密集型工作 withContext(Dispatchers.Default) { println("在默认线程上：${Thread.currentThread().name}") } // Unconfined：在调用者线程中启动，在挂起的地方恢复 withContext(Dispatchers.Unconfined) { println("无限制：${Thread.currentThread().name}") } } // 协程上下文元素有趣的上下文示例（）{ val 范围 = CoroutineScope( 调度员.Main + SupervisorJob() + 协程名称("MyCoroutine") + CoroutineExceptionHandler { _, 可抛出 -> println("捕获：$可抛出") } ）范围.启动{ println("上下文：$coroutineContext") } } // 继承上下文有趣的继承ContextExample() { CoroutineScope(Dispatchers.Main).launch { println("父级：${Thread.currentThread().name}") 发射{ // 继承Dispatchers.Main println("子进程：${Thread.currentThread().name}") } 启动（调度程序.IO）{ // 使用 IO 调度程序覆盖 println("覆盖: ${Thread.currentThread().name}") } } } // ThreadLocal 上下文元素 val threadLocalValue = ThreadLocal() 暂停有趣的 threadLocalExample() { threadLocalValue.set("初始") withContext(threadLocalValue.asContextElement("新值")) { println("在上下文中：${threadLocalValue.get()}") } println("上下文之后：${threadLocalValue.get()}") } // 自定义上下文元素数据类 UserId(val id: Int) : AbstractCoroutineContextElement(Key) { 伴随对象 Key : CoroutineContext.Key } 暂停有趣的customContextExample() { withContext(用户Id(42)) { val userId = coroutineContext[UserId] println("用户 ID：${userId?.id}") } } 调度程序确定哪个线程池执行协程。上下文元素由子协程继承并且可以被覆盖。 ## 反应流的流程 Flow 表示按需计算的冷值流，提供带有背压和转换运算符的反应式编程功能。 // 基本流程 fun numberFlow(): Flow = flow { for (i in 1..5) { 延迟(100) 发射(i) } } 暂停有趣的collectFlow() { numberFlow().collect { 值 -> println("已收到：$值") } } // 流程构建器有趣的流程构建器（）{ // flowOf：发出固定值 val 固定流量 = flowOf(1, 2, 3, 4, 5) // asFlow: 转换集合 val listFlow = listOf(1, 2, 3).asFlow() // flow: 自定义发射逻辑 val 自定义流量 = 流量 { 重复（3）{ 发出（它）延迟(100) } } } // 流变换暂停有趣的 flowTransformations() { 数字流() .map { 它 * 2 } .filter { it > 5 } .采取(3) .collect { println(it) } } // 流合并暂停有趣的combineFlows() { val flow1 = flowOf(1, 2, 3) val flow2 = flowOf("A", "B", "C") // zip: 对元素 flow1.zip(flow2) { 数字，字母 -> “$数字$字母” }.collect { println(it) } // 合并：每个最新的 flow1.combine(flow2) { 数字，字母 -> “$数字$字母” }.collect { println(it) } } // 流程异常处理暂停乐趣 flowExceptionHandling() { 流量{ 发出(1) 发射(2) 抛出异常（“错误！”） }.catch { e -> println("捕获：${e.message}") 发出(-1) }.collect { println(it) } } // StateFlow 和 SharedFlow 类数据存储库 { private val _users = MutableStateFlow>(emptyList()) val 用户：StateFlow> = _users 私有 val _events = MutableSharedFlow() val 事件：SharedFlow = _events 暂停有趣的 loadUsers() { val 已加载 = fetchUsers() _users.value = 已加载 } 暂停乐趣emitEvent（事件：事件）{ _events.emit(事件) } 私人挂起乐趣 fetchUsers(): List { 延迟(100) 返回列表（用户（1，“爱丽丝”）） } } 数据类事件（val类型：String） // 流向不同的调度程序暂停有趣的 flowWithContext() { 流量{ 发出(1) 发射(2) } .flowOn(调度程序.IO) .collect { 值 -> // 根据调用者的上下文收集 println("值：$值") } } // 热流的通道流有趣的通道流示例（）=通道流{ 发射{ 重复（3）{ 发送（它）延迟(100) } } 发射{ 重复（3）{ 发送（它 * 10）延迟(150) } } } 流是冷的——它在收集之前不会执行。 StateFlow 保存状态，SharedFlow 广播事件，channelFlow 支持并发发射。 ## 沟通渠道通道提供了用于在协程之间发送和接收值的通信原语，类似于 BlockingQueue 但暂停。 // 基本通道使用暂停有趣的channelExample() { val 通道 = Channel() 发射{ 对于 (x in 1..5) { 通道.发送(x) } 通道.close() } 对于（通道中的 y）{ println("已收到：$y") } } // 缓冲通道有趣的 bufferedChannelExample() { val 通道 = Channel(容量 = 4) GlobalScope.launch { for (x in 1..10) { println("正在发送$x") 通道.发送(x) } 通道.close() } GlobalScope.launch { 延迟(1000) 对于（通道中的 y）{ println("已收到：$y") 延迟(200) } } } // 通道类型有趣的通道类型() { // Rendezvous：无缓冲区，发送方暂停直到接收方 val rendezvous = Channel() // Buffered：指定容量 val 缓冲 = Channel(10) // Unlimited：无限缓冲 val 无限制 = Channel(Channel.UNLIMITED) // 合并：仅保留最新值 val conlated = Channel(Channel.CONFLATED) } // 生成构建器有趣的 ProduceNumbers(): ReceiveChannel = GlobalScope.product { 对于 (x in 1..5) { 发送（x * x）延迟(100) } } 暂停有趣的 ConsumerNumbers() { val 通道 = ProduceNumbers() 通道.consumeEach { println(it) } } // 多个消费者暂停有趣的 multipleConsumers() { val 通道 = Channel() 重复(3) { id -> 发射{ for（通道中的值）{ println("收到消费者$id：$value") } } } 重复（10）{ 频道.发送(it) 延迟(100) } 通道.close() } // 选择多个通道的表达式暂停有趣的 selectExample() { val 通道1 = 生产 { 发送("A") } val 通道2 = 生产 { 发送("B") } 选择<单位> { Channel1.onReceive { 值 -> println("来自通道1：$value") } Channel2.onReceive { 值 -> println("来自通道2：$value") } } } fun CoroutineScope.product(block: suspend () -> Unit): ReceiveChannel { 返回产品{ 块（） } } 通道支持扇出（多个消费者）、扇入（多个生产者）和管道模式以进行并发数据处理。 ## 取消和异常处理协程支持协作取消和结构化异常处理，以确保正确的资源清理和错误传播。 // 基本取消暂停乐趣cancelExample() { val job = GlobalScope.launch { 重复（1000）{我-> println("工作中：$i") 延迟(500) } } 延迟(2000) println("正在取消...") 作业.取消() 工作.加入() println("取消") } // 检查是否取消暂停有趣的 checkCancellation() { GlobalScope.launch { 变量我 = 0 while (isActive) { println("计算：${i++}") } } } // 不可取消的工作暂停乐趣 nonCancellableCleanup() { val job = GlobalScope.launch { 尝试{ 重复（1000）{ 延迟(500) } }最后{ withContext(不可取消) { println("在不可取消的上下文中进行清理") 延迟(1000) println("清理完成") } } } 延迟(1000) job.cancelAndJoin() } // 超时处理暂停有趣的 timeoutExample() { 尝试{ 超时（1000）{ 重复（100）{ 延迟(100) println("工作中...") } } } catch (e: TimeoutCancellationException) { println("超时") } } // 协程中的异常处理暂停有趣的异常处理示例（）{ val handler = CoroutineExceptionHandler { _, 异常 -> println("捕获：$异常") } GlobalScope.launch(处理程序) { 抛出异常（“协程失败”） } 延迟(100) } // 结构化异常处理挂起有趣的 StructuredExceptions() = coroutineScope { val job1 = 启动 { 延迟(100) 抛出异常（“作业 1 失败”） } val job2 = 启动 { 延迟(200) println("作业2完成") } // 当 job1 失败时 job2 被取消 } // 独立故障的 SupervisorScope 暂停乐趣supervisorExample（）=supervisorScope { val job1 = 启动 { 延迟(100) 抛出异常（“作业 1 失败”） } val job2 = 启动 { 延迟(200) println("作业2完成") } // 即使 job1 失败，job2 也会继续 } // 使用 try-catch 处理错误暂停有趣的错误处理（）{ 协程作用域 { 发射{ 尝试{ 获取用户(1) } catch (e: 异常) { println("错误：${e.message}") } } } } 取消是合作性的——协程必须检查“isActive”或调用挂起函数。异常处理遵循结构化并发边界。 ## 最佳实践 1. **使用具有作用域的结构化并发**来确保协程得到正确管理并在不再需要时取消 2. **为工作类型选择适当的调度程序**：IO 用于阻塞操作，Default 用于 CPU 工作，Main 用于 UI 更新 3. **通过在循环中检查 isActive 并使用支持取消的挂起函数来协作处理取消** 4. 对于反应流，**优先选择 Flow 而不是回调**，以获得背压、运算符和结构化生命周期管理 5. **使用supervisorScope进行独立操作**，防止一次失败取消不相关的协程 6. **除了真正的应用程序范围的工作之外，避免使用 GlobalScope** 以防止泄漏并保持结构化并发优势 7. **应用withContext进行调度程序切换**而不是启动新的协程来维护结构并减少开销 8. **使用 try-catch 或 CoroutineExceptionHandler 显式处理异常**，以防止静默失败 9. **使用 StateFlow 表示状态**，使用 SharedFlow 表示事件，以提供具有适当生命周期感知的可观察流 10. **使用 TestCoroutineDispatcher 测试协程**以控制时间并确保确定性测试执行 ## 常见陷阱 1. **使用 GlobalScope 进行作用域工作** 当协程的寿命超过其相关上下文（如活动或视图模型）时会导致内存泄漏 2. **使用 Thread.sleep 阻塞协程内部**或阻塞 I/O 达不到目的，并且会耗尽线程池 3. **在长时间运行的循环中不处理取消**会导致协程在取消后继续执行 4. **在调用其他挂起函数的函数上忘记挂起修饰符**会导致编译错误 5. **捕获 CancellationException** 并且不重新抛出它会阻止结构化并发中正确的取消传播 6. **使用delay(0)来yield**不如显式调用yield()进行协作多任务清晰 7. **不必要地创建太多协程**会降低性能；尽可能进行批量或节流操作 8. **不使用 withContext 进行调度程序切换**，而是启动不必要的子协程增加了复杂性 9. **假设启动后立即执行**；在调度程序有容量之前协程可能不会启动 10. **错误地混合回调和协程**，没有适当的桥接会产生竞争条件和泄漏 ## 何时使用此技能在构建 Android 应用程序以进行异步操作（例如网络调用、数据库查询或任何不应阻塞主线程的 I/O 密集型工作）时，请使用 Kotlin 协程。使用 Ktor 或 Spring Boot 在服务器端 Kotlin 应用程序中应用协程，以高效处理并发请求，而无需每个请求的线程开销。在 MVVM 架构中使用 Flow 进行反应式流，通过生命周期感知取代 LiveData 或 RxJava 进行状态管理和事件传播。利用结构化并发来协调多个异步操作，确保在离开屏幕或关闭连接时正确取消。将通道用于生产者-消费者模式、管道或任何需要并发协程之间显式通信的场景。 ## 资源 * [Kotlin 协程指南]() * [Android 上的协程]() * [流程文档]() * [KotlinConf 协程讲座]() * [协程最佳实践]()

Kotlin 协程

包含技能