Kotlin如何实现返回用户后异步执行AWS、MongoDB、Kafka任务

你要的「返回主结果后不阻塞执行后置IO任务」的场景，核心是用结构化并发的协程作用域启动独立后台任务，避免协程泄漏和未捕获异常崩溃，不要直接用GlobalScope，也不要用runBlocking（会阻塞当前执行线程，完全失去异步意义）。

第一步：定义应用级后台协程作用域 #

首先创建和应用生命周期绑定的协程作用域，替代无约束的GlobalScope，统一处理异常、调度器和生命周期：

// 应用启动时初始化，应用停机时调用cancel()终止所有未完成任务即可
// 比如Spring环境可以声明为单例Bean，在@PreDestroy生命周期方法里执行关闭
private val backgroundTaskScope = CoroutineScope(
    SupervisorJob() + // 子任务异常不影响其他任务执行，不会牵连整个作用域
    Dispatchers.IO + // 网络请求、数据库操作等IO密集型任务统一用IO调度器
    CoroutineExceptionHandler { _, throwable ->
        // 统一处理异步任务异常：打错误日志、上报监控告警即可，不要把异常抛出
        log.error("后台异步任务执行异常", throwable)
    }
)

第二步：改造doStuff函数 #

直接在前置逻辑执行完成、拿到要返回的user对象后，启动协程执行异步任务，不需要等待任务完成就直接返回结果：

fun doStuff(newUser: User) : ReturnType {
    // 执行原有前置逻辑
    val user = runPreLogic(newUser)

    // 启动非阻塞后台任务，不会阻塞当前执行流程
    backgroundTaskScope.launch {
        // 三个IO任务无依赖的话直接并行执行，比串行执行效率高很多
        coroutineScope {
            val awsTask = async { runAwsOperations(user) }
            val mongoTask = async { runMongoInsert(user) }
            val kafkaTask = async { runKafkaSend(user) }
            // 等待所有任务执行完成，异常会被外层的异常处理器统一捕获
            awsTask.await()
            mongoTask.await()
            kafkaTask.await()
        }
    }

    // 立即返回结果，无需等待后台任务
    return user
}

特殊场景适配 #

如果你的doStuff本身是suspend挂起函数，运行在Web请求等短生命周期协程上下文里，可以根据任务的生命周期要求选实现：

• 如果后台任务不需要等请求返回，就算请求断开、客户端取消也要跑完：直接用上面的应用级backgroundTaskScope启动任务即可
• 如果后台任务可以跟随请求生命周期终止（比如客户端断开连接就没必要继续执行）：用当前上下文的协程作用域启动即可，记得加SupervisorJob避免单个任务失败影响请求主流程：

suspend fun doStuff(newUser: User) : ReturnType {
    val user = runPreLogic(newUser)

    coroutineScope {
        launch(SupervisorJob() + CoroutineExceptionHandler { _, e -> log.error("请求内异步任务失败", e) }) {
            listOf(
                async { runAwsOperations(user) },
                async { runMongoInsert(user) },
                async { runKafkaSend(user) }
            ).awaitAll()
        }
    }

    return user
}

注意事项 #

• 不要在异步协程块里修改要返回的user对象，避免并发读写问题，传入只读的用户数据副本最稳妥
• 不要把Dispatchers.IO换成Dispatchers.Default：后者是给CPU密集型计算任务用的，IO调度器才适配网络、数据库这类阻塞操作
• 如果是计费、订单等高可靠性场景，不要完全依赖内存中的协程执行任务：先把待执行的异步任务持久化到数据库，再用调度组件捞取执行，避免应用重启、进程崩溃导致任务丢失
• 不要在协程块里写阻塞线程的代码，如果必须调用阻塞SDK，记得用withContext(Dispatchers.IO)包裹

内容的提问来源于stack exchange，提问作者Guilherme Melo