如何优化 Kotlin 中的 n 个异步并发调用

Question

我在 Kotlin 中有以下代码，很好奇是否有一种方法可以进一步优化它，与我认为有些等效的 Python 实现相比？

import java.net.URI
import java.net.http.HttpClient
import java.net.http.HttpRequest
import java.net.http.HttpResponse
import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

fun main() = runBlocking {
    val timeElapsed = measureTimeMillis {
        val deferreds: List<Deferred<String>> = (1..7).map {
            async {
                val client = HttpClient.newBuilder().build();
                val request = HttpRequest.newBuilder()
                    .uri(URI.create("URL_WITH_INCREASING_PATH_PARAMETER/$it"))
                    .build();

                val response = client.send(request, HttpResponse.BodyHandlers.ofString());
                response.body()
            }
        }

        val data = deferreds.awaitAll()
        println("$data")
    }

    println("time elapsed $timeElapsed")
}

// time elapsed 2126

import aiohttp
import asyncio
import time

start_time = time.time()

async def get_data(session, url):
    async with session.get(url) as response:
        data = await response.json()
        return data

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = []
        for number in range(1, 7):
            url = f'URL_WITH_INCREASING_PATH_PARAMETER/{number}'
            tasks.append(asyncio.ensure_future(get_data(session, url)))
        data_list = await asyncio.gather(*tasks)
        print(data_list)

asyncio.run(main())
print('time elapsed %s' % (time.time() - start_time))

# time elapsed 0.7383

Answer 1

你正在使用 JDK 11 的

HttpClient.send()

方法，它是阻塞的。此外，您正在使用

runBlocking

而不覆盖其调度程序，因此您将获得一个单线程事件循环作为调度程序。

这意味着这里没有任何并行运行。

您可以做的第一件事是使用

Dispatchers.IO

运行这些阻塞 IO 调用：

fun main() = runBlocking(Dispatchers.IO) { // switching to IO dispatcher
    val client = HttpClient.newBuilder().build();
    val timeElapsed = measureTimeMillis {
        val deferreds = List(7) {
            async {
                val request = HttpRequest.newBuilder()
                    .uri(URI.create("URL_WITH_INCREASING_PATH_PARAMETER/$it"))
                    .build();

                val response = client.send(request, HttpResponse.BodyHandlers.ofString());
                response.body()
            }
        }

        val data = deferreds.awaitAll()
        println("$data")
    }

    println("time elapsed $timeElapsed")
}

但这是使用了比必要更多的线程，因为很可能 HTTP 客户端已经有自己的线程池了。

更好的选择是使用

send

方法的实际异步变体：

sendAsync

。它返回一个

CompletableFuture

，您可以使用

CompletableFuture.await()

以暂停的方式等待