如何使用协程异步处理数据？

Question

我需要执行 4 个并行请求。这是我的代码：

suspend fun fetchAsyncData() = coroutineScope {
    val first = async { repository.taskFirst() }
    val second = async { repository.taskSecond() }
    val third = async { repository.taskThird() }
    val fourth = async { repository.taskFourth() }

    val firstResult = first.await()
    val secondResult = second.await()
    val thirdResult = third.await()
    val fourthResult = fourth.await()
    
}

问题在于，使用这种方法，请求是并行执行的，但我可以同时得到答案。也就是说，如果有些请求将执行45秒，有些请求将执行3秒，那么我只有在45秒后才能处理我的请求结果。 我的任务是，一旦收到第一个请求的答案，就将其传递给视图模型，以便视图模型可以在片段中显示这段数据。此外，一旦收到对请求的另一响应，就再传输一个数据，依此类推。

这该怎么办，请帮帮我？

Answer 1

也许这样的东西适合你：

fun getRepoFlow() = channelFlow {

        coroutineScope {
            launch { send(async { "1" }.await()) }
            launch { send(async { delay(1000L); "2" }.await()) }
            launch { send(async { "3" }.await()) }
        }

    }

Answer 2

如果所有任务返回相同的结果，那么您可以创建一个流并向其异步发送数据。这是一个例子：

fun main() {
    // a flow object inside your viewModel
    val flow: MutableStateFlow<Data?> = MutableStateFlow(null)

    // observe the flow. e.g. For android in Activity's lifecycle
    CoroutineScope(Dispatchers.Unconfined).launch {
        flow.collect {
            println(it)
        }
    }

    // get data from viewModel
    CoroutineScope(Dispatchers.Unconfined).launch {
        launch {
            flow.value = task1()
        }

        launch {
            flow.value = task2()
        }
    }

    // app's lifespan. Not needed in practical applications.
    while (true) {}
}

suspend fun task1(): Data {
    delay(2000)
    return Data(1, Date())
}

suspend fun task2(): Data {
    delay(1000)
    return Data(2, Date())
}

data class Data(val d: Int, val time: Date)

但是如果

tasks

返回不同的数据，那么您可以创建多个

flow

并在每个任务返回时发送数据。

Answer 3

通过这种方法，您可以处理不同类型的响应和失败案例。

       sealed class ResponseState {
            data class Content<T>(val data: T) : ResponseState()
            // handle network and server error state
            //data class error(e : Throwable): ResponseState()

        }

        val flow: MutableStateFlow<ResponseState?> = MutableStateFlow(null)
        coroutineScope {
            launch {
                val res = repository.taskFirst() // return type a
                flow.value = ResponseState.Content<a>(res)
            }
            launch {
                val res = repository.taskSecond()// return type b
                flow.value = ResponseState.Content<b>(res)
            }
            launch {
                val res = repository.taskThird() // return type c
                flow.value = ResponseState.Content<c>(res)
            }
            launch {
                val res = repository.taskFourth() // return type d
                flow.value = ResponseState.Content<d>(res)
            }
        }
        // here is an example with flow, you can use live data as well
        CoroutineScope(Dispatchers.Main).launch {
            flow.collect {
                when (it) {
                    is ResponseState.Content<*> -> {
                        // check response type
                        when (it.data) {
                            is a -> {
                                //handle taskFirst response
                            }

                            is b -> {
                                //handle taskSecond response
                            }

                            is c -> {
                                //handle taskThird response
                            }

                            is d -> {
                                //handle taskFourth response
                            }

                            else -> {}
                        }
                    }
                }
            }
        }