amazon-s3 - 通过 Kotlin 协程并发 S3 文件上传

Question

我需要将许多文件上传到 S3，依次完成该工作需要数小时。这正是 Kotlin 的新协程所擅长的，所以我想给它们第一次尝试，而不是再次摆弄一些基于线程的执行服务。

这是我的(简化)代码:

fun upload(superTiles: Map<Int, Map<Int, SuperTile>>) = runBlocking {
    val s3 = AmazonS3ClientBuilder.standard().withRegion("eu-west-1").build()
    for ((x, ys) in superTiles) {
        val jobs = mutableListOf<Deferred<Any>>()
        for ((y, superTile) in ys) {
            val job = async(CommonPool) {
                uploadTile(s3, x, y, superTile)
            }
            jobs.add(job)
        }
        jobs.map { it.await() }
    }
}

suspend fun uploadTile(s3: AmazonS3, x: Int, y: Int, superTile: SuperTile) {
    val json: String = "{}"
    val key = "$s3Prefix/x4/$z/$x/$y.json"
    s3.putObject(PutObjectRequest("my_bucket", ByteArrayInputStream(json.toByteArray()), metadata))
}

问题:代码仍然很慢，并且日志显示请求仍然按顺序执行:在创建下一个作业之前完成了一项作业。仅在极少数情况下(十分之 1)我会看到作业同时运行。

为什么代码没有运行得更快/并发？我该怎么办？

Answer 1

Kotlin 协程在您使用异步 API 时表现出色，而 AmazonS3.putObject您使用的 API 是一个老式的阻塞同步 API，因此您只能获得与 CommonPool 中的线程数一样多的并发上传。你正在使用的。标记您的 uploadTile 没有任何值(value)功能与 suspend修改，因为它在其主体中不使用任何挂起函数。

在上传任务中获得更多吞吐量的第一步是开始为此使用异步 API。我建议看 Amazon S3 TransferManager为了那个钱包。看看这是否首先解决了您的问题。

Kotlin 协程旨在帮助您将异步 API 组合成易于使用的逻辑工作流。例如，适配 TransferManager 的异步 API 很简单。通过编写以下扩展函数与协程一起使用:

suspend fun Upload.await(): UploadResult = suspendCancellableCoroutine { cont ->
    addProgressListener {
        if (isDone) {
            // we know it should not actually wait when done
            try { cont.resume(waitForUploadResult()) }
            catch (e: Throwable) { cont.resumeWithException(e) }
        }
    }
    cont.invokeOnCompletion { abort() }
}

此扩展使您能够编写与 TransferManager 一起使用的非常流畅的代码。你可以重写你的 uploadTile使用的函数 TransferManager而不是使用阻塞 AmazonS3界面:

suspend fun uploadTile(tm: TransferManager, x: Int, y: Int, superTile: SuperTile) {
    val json: String = "{}"
    val key = "$s3Prefix/x4/$z/$x/$y.json"
    tm.upload(PutObjectRequest("my_bucket", ByteArrayInputStream(json.toByteArray()), metadata))
        .await()
}

注意，这个新版本的 uploadTile使用挂起函数 await这是上面定义的。