Alpakka S3 连接器流不会处理负载，抛出 akka.stream.BufferOverflowException

我有一个akka-http服务，我正在尝试使用alpakka s3连接器上传文件。以前，我使用临时文件，然后使用 Amazon 开发工具包上传。这种方法需要对 Amazon 开发工具包进行一些调整，使其更像 scala，但它甚至可以一次处理 1000 个请求。吞吐量并不惊人，但所有请求最终都通过了。以下是更改前的代码，没有alpakka：

'''

path("uploadfile") {
withRequestTimeout(20.seconds) {
storeUploadedFile("csv", tempDestination) {
case (metadata, file) =>
val uploadFuture = upload(file, file.toPath.getFileName.toString)
onComplete(uploadFuture) {
case Success(_) => complete(StatusCodes.OK)
case Failure(_) => complete(StatusCodes.FailedDependency)
}
}
}
}
}
case class S3UploaderException(msg: String) extends Exception(msg)
def upload(file: File, key: String): Future[String] = {
val s3Client = AmazonS3ClientBuilder.standard()
.withCredentials(new DefaultAWSCredentialsProviderChain())
.withRegion(Regions.EU_WEST_3)
.build()
val promise = Promise[String]()
val listener = new ProgressListener() {
override def progressChanged(progressEvent: ProgressEvent): Unit = {
(progressEvent.getEventType: @unchecked) match {
case ProgressEventType.TRANSFER_FAILED_EVENT => promise.failure(S3UploaderException(s"Uploading a file with a key: $key"))
case ProgressEventType.TRANSFER_COMPLETED_EVENT |
ProgressEventType.TRANSFER_CANCELED_EVENT => promise.success(key)
}
}
}
val request = new PutObjectRequest("S3_BUCKET", key, file)
request.setGeneralProgressListener(listener)
s3Client.putObject(request)
promise.future
}

'''

当我将其更改为使用 alpakka 连接器时，代码看起来更好，因为我们可以将ByteSource和 alpakkaSink连接在一起。但是，这种方法无法处理如此大的负载。当我一次执行 1000 个请求(10 kb 文件(时，只有不到 10% 的请求通过，其余请求失败并出现异常：

akka.stream.alpakka.s3.impl.FailedUpload：已超出配置最大打开请求值为 [32]。这意味着请求队列这个池 (HostConnectionPoolSetup(bargain-test.s3-eu-west-3.amazonaws.com，443，ConnectionPoolSetup(ConnectionPoolSettings(4,0,5,32,1,30 秒，ClientConnectionSettings(Some(User-Agent： akka-http/10.1.3(，10 秒，1 分钟，512，无，WebSocketSettings(，ping，Duration.Inf，akka.http.impl.settings.WebSocketSettingsImpl$$$Lambda$4787/1279590204@4d809f4c(，List((，ParserSettings(2048,16,64,64,8192,64,8388608,256,1048576，Strict，RFC6265，true，Set((，Full，Error，Map(If-Range -> 0，如果修改以来 -> 0，如果未修改以来 -> 0，默认值 -> 12，内容-MD5 -> 0，日期 -> 0，如果匹配 -> 0，如果无匹配 -> 0， User-Agent ->32(，false，true，akka.util.ConstantFun$$$Lambda$4534/1539966798@69c23cd4，akka.util.ConstantFun$$$Lambda$4534/1539966798@69c23cd4，akka.util.ConstantFun$$$Lambda$4535/297570074@6b426c59(，None，TCPTransport(，New，1 第二(，akka.http.scaladsl.HttpsConnectionContext@7e0f3726，akka.event.MarkerLoggingAdapter@74f3a78b((( 已完全填满，因为池当前未处理请求速度足以处理传入的请求负载。请重试稍后的请求。看 http://doc.akka.io/docs/akka-http/current/scala/http/client-side/pool-overflow.html 了解更多信息。

以下是加特林测试的摘要：

----响应时间分布---------------------------------------- 吨 <800 毫秒 0 ( 0%(
800 毫秒<吨>
吨> 1200 毫秒 90 ( 9%(
失败 910 ( 91%(

当我同时执行 100 个请求时，其中一半失败。所以，仍然接近令人满意。

这是一个新代码： ```

path("uploadfile") {
withRequestTimeout(20.seconds) {
extractRequestContext { ctx =>
implicit val materializer = ctx.materializer
extractActorSystem { actorSystem =>
fileUpload("csv") {
case (metadata, byteSource) =>
val uploadFuture = byteSource.runWith(S3Uploader.sink("s3FileKey")(actorSystem, materializer))
onComplete(uploadFuture) {
case Success(_) => complete(StatusCodes.OK)
case Failure(_) => complete(StatusCodes.FailedDependency)
}
}            
}
}
}
}
def sink(s3Key: String)(implicit as: ActorSystem, m: Materializer) = {
val regionProvider = new AwsRegionProvider {
def getRegion: String = Regions.EU_WEST_3.getName
}
val settings = new S3Settings(MemoryBufferType, None, new DefaultAWSCredentialsProviderChain(), regionProvider, false, None, ListBucketVersion2)
val s3Client = new S3Client(settings)(as, m)
s3Client.multipartUpload("S3_BUCKET", s3Key)
}

'''

可以在此处查看两个端点的完整代码

我有几个问题。

1( 这是一个功能吗？这就是我们可以称之为背压吗？

2(如果我希望这段代码的行为类似于带有临时文件的旧方法(没有失败的请求，并且所有请求都在某个时候完成(，我该怎么办？我试图为流实现一个队列(链接到下面的源代码(，但这没有区别。代码可以在这里看到。

(*免责声明*我仍然是一个scala新手，试图快速理解akka流并找到解决问题的方法。此代码中很有可能存在一些简单的错误。* 免责声明 *(

这是一个背压功能。

Exceeded configured max-open-requests value of [32]在配置中，max-open-requests默认设置为 32。流式处理用于处理大量数据，而不是每秒处理许多请求。

Akka开发人员不得不为max-open-requests放一些东西。他们出于某种原因选择32个肯定。他们不知道它将用于什么。它可以一次发送 1000 个 32KB 文件或 1000 个 1GB 文件吗？他们不知道。但他们仍然希望确保默认情况下(80% 的人可能使用默认值(应用程序将得到优雅和安全的处理。所以他们不得不限制处理能力。

您要求"现在"执行 1000 个文件，但我很确定 AWS 没有同时发送 1000 个文件，而是使用了一些队列，如果您有很多小文件要上传，这对您来说可能也是一个很好的案例。

但是完全可以根据您的情况进行调整！如果您知道您的机器和目标将处理更多的同时连接，则可以将数字更改为更高的值。

此外，对于许多 HTTP 调用，请使用缓存的主机连接池。

相关内容

最新更新

热门标签：