5

我需要将许多文件上传到 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))
}

问题：代码仍然很慢，日志显示请求仍然按顺序执行：一个作业在创建下一个作业之前完成。只有在极少数情况下（十分之一）我看到作业同时运行。

为什么代码不能更快/同时运行？我能做些什么呢？

4

1 回答 1

13

当您使用异步API 时，Kotlin 协程表现出色，而AmazonS3.putObject您使用的 API 是老式的阻塞同步 API，因此您获得的并发上传数与CommonPool您使用的线程数一样多。用修改标记你的uploadTile函数没有任何价值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

于 2017-07-19T09:35:01.313 回答