使用Akka HTTP+Alpakka S3传大文件遇ParsingException求助

我来帮你梳理下这个问题的根源和对应的解决办法，你的场景是小文件上传正常，但90MB的大文件会抛出Unexpected end of multipart entity异常，且文件仅上传了45-50MB左右，虽然收到成功提示但实际并未完成，核心问题出在Akka HTTP的multipart解析配置、流处理的完整性，以及Alpakka S3的分段上传设置上。

问题根源 #

1. Akka HTTP默认的multipart解析限制：即使你用了withoutSizeLimit，Akka HTTP的multipart解析器默认还有缓冲区大小、超时时间的限制，大文件上传时如果解析超时或者缓冲区不足，会导致数据流被提前截断，抛出Unexpected end of multipart entity。
2. 流处理的完整性缺失：你的代码中直接启动上传流后就等待结果，但Akka HTTP的请求处理周期可能在流未完全消费完就提前返回响应，导致数据流被终止，进而文件上传不完整。
3. Alpakka S3分段上传的默认配置：默认的分段大小较小，大文件会被分成大量分段上传，增加了出错概率。

具体解决步骤 #

1. 调整Akka HTTP的multipart解析配置 #

在你的application.conf中添加以下配置，放宽解析限制并延长超时：

akka.http.parsing.multipart {
  max-part-size = 100MB  # 设置为比你最大上传文件稍大的值
  max-content-length = 100MB
  chunk-streaming-timeout = 60s  # 延长超时时间，避免大文件上传时超时截断
}

2. 确保数据流被完整消费 #

修改路由代码，添加流终止监控，确保整个文件流被完全消费后再返回响应：

def uploadLargeFile: Route = post {
  path("import" / "file") {
    extractMaterializer { implicit materializer =>
      extractActorSystem { implicit system =>
        withoutSizeLimit {
          fileUpload("file") { case (metadata, byteSource) =>
            logger.info(s"Request received to import large file: ${metadata.fileName}")
            
            // 监控流的终止状态，确保完整消费
            val uploadFuture = byteSource
              .watchTermination() { (_, termination) =>
                termination.onComplete {
                  case Success(_) => logger.info(s"File stream for ${metadata.fileName} fully consumed")
                  case Failure(ex) => logger.error(s"File stream for ${metadata.fileName} terminated early", ex)
                }
                termination
              }
              .runWith(s3Utility.sink(metadata))

            onComplete(uploadFuture) {
              case Success(result) =>
                logger.info(s"Successfully uploaded file: ${result.location}")
                complete(StatusCodes.OK, s"File ${metadata.fileName} uploaded successfully")
              case Failure(ex) =>
                logger.error("Error in uploading file", ex)
                complete(StatusCodes.InternalServerError, s"Upload failed: ${ex.getMessage}")
            }
          }
        }
      }
    }
  }
}

3. 优化Alpakka S3的分段上传设置 #

调整分段大小，减少分段数量，提升大文件上传的稳定性：

class S3Utility(implicit as: ActorSystem, m: Materializer) {
  private val bucketName = "test"
  // 设置分段大小为10MB（可根据你的网络情况调整，建议5-20MB）
  private val multipartSettings = MultipartUploadSettings(maxPartSize = 10 * 1024 * 1024)

  def sink(fileInfo: FileInfo): Sink[ByteString, Future[MultipartUploadResult]] = {
    val fileName = fileInfo.fileName
    S3.multipartUpload(bucketName, fileName, settings = multipartSettings)
  }
}

4. 检查客户端上传配置 #

如果是前端或其他客户端发起的上传，也要确保客户端没有设置过短的超时时间或大小限制，比如浏览器的默认超时、HTTP客户端的请求超时配置，确保能完整发送整个大文件流。

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