如何用Kotlin挂起协程处理回调的多次数据更新？

你的核心问题在于挂起函数只能单次返回结果，而onData会被多次调用，直接用suspendCancellableCoroutine必然会因为重复恢复协程崩溃。正确的做法是使用Flow来承载多次发射的数据，推荐用Kotlin协程的callbackFlow构建器——它专门用于将基于回调的API转换为冷流。

修改后的代码示例 #

override fun foo(p0: SomeType): Flow<Long> = callbackFlow {
    val process = object : CallbackFunction() {
        override fun onData(bytes: Long) {
            // 尝试向流发射数据，若流已取消则静默失败
            trySend(bytes).isSuccess
        }

        override fun onComplete() {
            // 流完成，关闭通道
            close()
        }

        override fun onFailure(exp: Exception) {
            // 发送异常并关闭流
            close(exp)
        }
    }

    // 当Flow被取消时，清理资源（比如停止process）
    awaitClose {
        // 假设process有cancel方法，根据实际API调整
        // process.cancel()
    }
}

关键说明 #

• 返回类型变更：将原来的suspend fun foo(): String改为fun foo(): Flow<Long>，因为我们要发射多次Long类型的onData结果，而非单次返回值。
• callbackFlow的作用：它内部使用通道（Channel）来传递数据，天然支持多值发射，同时处理协程的取消逻辑。
• trySend的使用：相比send，trySend不会挂起，适合在回调线程中调用；isSuccess可以判断是否发送成功（比如流已取消时会失败）。
• awaitClose的必要性：在协程取消时执行资源清理，避免内存泄漏。如果你的process有取消/停止方法，一定要在这里调用。
• 异常与完成处理：onComplete调用close()结束流；onFailure调用close(exp)将异常传递给流的收集方。

如何收集这个Flow #

在调用方，你可以用协程收集流中的数据：

lifecycleScope.launch {
    foo(someTypeInstance).collect { bytes ->
        // 处理每次onData返回的bytes
        println("Received bytes: $bytes")
    }
}

内容的提问来源于stack exchange，提问作者SURYA N