实现一个并发数据加载器并处理异常

题目

实现一个并发数据加载器并处理异常

信息

• 类型：问答
• 难度：⭐⭐

考点

协程作用域管理, 结构化并发, 异常处理, 挂起函数组合

快速回答

实现要点：

• 使用 coroutineScope 创建独立作用域管理并发任务
• 通过 async 启动并发子协程
• 使用 SupervisorJob 实现子协程异常隔离
• 在父协程中集中处理 CancellationException 和常规异常
• 通过 awaitAll() 等待所有子协程完成

问题场景

需要从三个独立数据源并发加载用户信息、订单记录和库存数据，任一数据源加载失败不应影响其他请求，但需要收集所有异常信息。

解决方案

suspend fun loadAllData(): Result<Triple<User, List<Order>, Int>> {
    return coroutineScope {
        val userDeferred = async(SupervisorJob()) { loadUser() }
        val ordersDeferred = async(SupervisorJob()) { loadOrders() }
        val stockDeferred = async(SupervisorJob()) { loadStock() }

        val exceptions = mutableListOf<Throwable>()

        try {
            // 等待所有协程完成（包括失败）
            awaitAll(userDeferred, ordersDeferred, stockDeferred)
        } catch (e: CancellationException) {
            throw e  // 传播取消异常
        } catch (e: Exception) {
            exceptions.add(e)
        }

        // 收集子协程的独立异常
        listOf(userDeferred, ordersDeferred, stockDeferred).forEach { deferred ->
            if (deferred.isCancelled) {
                runCatching { deferred.await() }
                    .onFailure { exceptions.add(it) }
            }
        }

        if (exceptions.isNotEmpty()) {
            return@coroutineScope Result.failure(MultiException(exceptions))
        }

        Result.success(
            Triple(
                userDeferred.await(),
                ordersDeferred.await(),
                stockDeferred.await()
            )
        )
    }
}

// 模拟数据源
suspend fun loadUser(): User { /* ... */ }
suspend fun loadOrders(): List<Order> { /* ... */ }
suspend fun loadStock(): Int { /* ... */ }

class MultiException(val exceptions: List<Throwable>) : Exception()

核心原理

• 结构化并发：通过 coroutineScope 创建子作用域，确保所有子协程在返回前完成
• 异常隔离：SupervisorJob 使子协程异常互不影响
• 错误聚合：先捕获父协程异常，再单独处理每个子协程的异常状态
• 取消传播：优先处理 CancellationException 保证协程取消机制正常工作

最佳实践

• 为每个并发任务使用独立的 SupervisorJob
• 使用 awaitAll() 而非单独 await() 确保所有协程执行完毕
• 区分取消异常和业务异常处理逻辑
• 通过 isCancelled 检查协程状态避免重复捕获异常

常见错误

• 错误：直接使用 async { } 导致一个子协程异常取消所有兄弟协程
• 错误：在 try-catch 中单独 await() 导致未处理异常丢失
• 错误：忽略 CancellationException 破坏协程取消机制
• 错误：使用 GlobalScope 导致生命周期管理失控

扩展知识

• CoroutineExceptionHandler：全局异常处理，但无法解决并发异常聚合问题
• Result 封装：Kotlin 1.5+ 的 Result 类可简化错误处理
• async 启动模式：async(start = CoroutineStart.LAZY) 实现延迟执行
• Flow 方案：使用 flow { ... }.catch { ... } 组合异步流，但并发控制更复杂