Kotlin Async/Await：发送多GET请求并实时处理返回结果

我尝试向服务器发送多个GET请求，希望在每个请求的结果返回时立即进行处理，但当前代码是按顺序等待所有请求完成后再处理结果。相关代码和控制台输出如下：

原代码 #

suspend fun getRequests(value: Int): Int {
    val rnd = (100..3000L).random()
    println("Delay for $rnd") // simulate network time delay.
    delay(rnd)
    return value * 2
}

fun main() {
    runBlocking {
        launch(Dispatchers.IO) {
            val myList: ArrayList<Deferred<Int>> = ArrayList()
            for (i in 1..10) {
                println("GET request sent: no$i")
                myList.add(async { getRequests(i) })
            }
            myList.forEach { println(it.await()) } // process the results
        }
    }
}

原控制台输出 #

GET request sent: no1
GET request sent: no2
GET request sent: no3
GET request sent: no4
GET request sent: no5
GET request sent: no6
GET request sent: no7
Delay for 706
Delay for 487
Delay for 2894
Delay for 1857
Delay for 353
GET request sent: no8
Delay for 2216
GET request sent: no9
Delay for 546
GET request sent: no10
Delay for 1806
Delay for 1291
Delay for 2830
2
4
6
8
10
12
14
16
18
20

问题原因分析 #

你现在的写法之所以会按顺序输出结果，是因为你先把所有Deferred对象收集到列表中，然后按列表顺序逐个调用await()。哪怕某个请求已经提前完成，也得等列表中排在它前面的Deferred.await()执行完毕，才会处理它的结果——这就导致所有结果要等全部请求都完成后，才按原请求的顺序输出，而不是完成即处理。

解决方案：为每个请求单独启动结果处理协程 #

要实现“结果一返回就立即处理”，你需要在每个异步请求启动后，立刻为它的结果处理单独开一个协程，而不是等到所有请求都启动完再统一处理。

修改后的代码示例1（保留async）

suspend fun getRequests(value: Int): Int {
    val rnd = (100..3000L).random()
    println("Delay for $rnd") // simulate network time delay.
    delay(rnd)
    return value * 2
}

fun main() {
    runBlocking {
        launch(Dispatchers.IO) {
            for (i in 1..10) {
                println("GET request sent: no$i")
                // 每个async启动后，立即启动协程等待结果并处理
                launch {
                    val result = async { getRequests(i) }.await()
                    println("Processed result: $result")
                }
            }
        }
    }
}

修改后的代码示例2（更简洁，无需async）

如果不需要保留对异步任务的Deferred引用，直接在launch里调用suspend函数即可：

fun main() {
    runBlocking {
        launch(Dispatchers.IO) {
            repeat(10) { idx ->
                val requestNum = idx + 1
                println("GET request sent: no$requestNum")
                launch {
                    val result = getRequests(requestNum)
                    println("Processed result: $result")
                }
            }
        }
    }
}

预期输出效果

结果会按请求完成的先后顺序输出，比如：

GET request sent: no1
GET request sent: no2
GET request sent: no3
GET request sent: no4
GET request sent: no5
Delay for 353  // no5的延迟最短
GET request sent: no6
...
Processed result: 10  // no5的结果最先被处理
Processed result: 4   // no2的结果随后处理
Processed result: 18  // no9的结果接着处理
... 后续结果按完成顺序输出

进阶：等待所有任务处理完成 #

如果你需要在所有请求的结果都处理完后执行后续操作，可以收集每个处理协程的Job，最后调用joinAll()：

fun main() {
    runBlocking {
        launch(Dispatchers.IO) {
            val processingJobs = mutableListOf<Job>()
            
            for (i in 1..10) {
                println("GET request sent: no$i")
                val job = launch {
                    val result = getRequests(i)
                    println("Processed result: $result")
                }
                processingJobs.add(job)
            }
            
            processingJobs.joinAll() // 等待所有结果处理完成
            println("All requests have been processed successfully!")
        }
    }
}

内容的提问来源于stack exchange，提问作者jake wong