使用Akka从AWS S3向Facebook分片上传时的S3对象转ByteString问题

最近我正好在做这个需求，结合Facebook分片上传的API要求，用Akka配合Java AWS SDK实现了整个流程，下面是具体的思路和代码片段：

核心逻辑梳理 #

Facebook的分片上传机制是：根据文件总大小，每次上传后返回下一个需要上传的字节偏移量，我们需要精准从S3拉取对应范围的数据块，再异步提交给Facebook。Akka的异步流处理特性正好适配这种分阶段、需要持续状态更新的场景。

关键代码实现 #

1. 构建S3范围请求获取数据块 #

首先用Java AWS SDK创建指定字节范围的GetObjectRequest，拉取对应分片：

// 从业务状态中获取S3对象信息、Facebook返回的下一个偏移量
val objChunkReq = new GetObjectRequest(get.s3ObjId.bucketName, get.s3ObjId.key)
// 设置请求的字节范围：起始偏移是Facebook返回的fbUploadOffset，结束偏移是起始偏移+分片大小-1
// 注意：如果剩余字节不足分片大小，结束偏移要设为文件总大小-1，避免无效范围请求
val endOffset = Math.min(get.fbUploadOffset + chunkSize - 1, get.totalFileSize - 1)
objChunkReq.setRange(get.fbUploadOffset, endOffset)

2. 将S3数据流转为Akka Stream Source #

为了适配Akka的异步处理，把S3返回的输入流转换成Akka的Source：

import akka.stream.scaladsl.Source
import akka.util.ByteString
import com.amazonaws.services.s3.AmazonS3ClientBuilder

val s3Client = AmazonS3ClientBuilder.defaultClient()
val s3Object = s3Client.getObject(objChunkReq)
val inputStream = s3Object.getObjectContent()

// 将输入流转为Akka Stream的Source，方便后续异步上传
val chunkSource = Source.fromInputStream(inputStream)
  .map(ByteString(_))
  .watchTermination() { (_, done) =>
    // 确保流结束后关闭S3的输入流，避免资源泄漏
    done.onComplete(_ => inputStream.close())(system.dispatcher)
  }

3. 异步上传分片到Facebook #

用Akka HTTP构建上传请求，把分片数据提交给Facebook的分片上传API，然后解析响应获取新的偏移量：

import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, Multipart, RequestEntity}

// 构建Facebook要求的表单数据，包含offset、file_size等必填参数
val formData = Multipart.FormData(
  Multipart.FormData.BodyPart.Strict("offset", get.fbUploadOffset.toString),
  Multipart.FormData.BodyPart.Strict("file_size", get.totalFileSize.toString),
  // 把分片数据作为文件部分上传
  Multipart.FormData.BodyPart.fromSource("file", chunkSource)
)

val uploadRequest = HttpRequest(
  method = HttpMethods.POST,
  uri = s"https://graph.facebook.com/v18.0/${get.fbUploadSessionId}/chunk",
  entity = formData.toEntity
)

// 发送异步请求并处理响应
Http().singleRequest(uploadRequest).map { response =>
  // 解析Facebook返回的响应，提取新的偏移量nextOffset
  // 更新上传会话状态，继续下一轮分片上传，直到nextOffset等于文件总大小（上传完成）
}.recover {
  // 处理上传失败的情况，比如重试拉取S3分片并重传
  case ex: Exception => // 重试逻辑或错误处理
}

实践中的注意事项 #

• 边界处理：一定要注意最后一个分片的范围计算，避免请求超出文件总大小的无效范围，导致S3返回416错误。
• 资源管理：S3的ObjectContent必须在使用后关闭，这里用watchTermination确保流结束后自动关闭输入流。
• 状态管理：可以用Akka的Actor或者FSM来维护上传会话的状态（当前偏移量、上传会话ID、文件总大小等），让多轮分片上传的逻辑更清晰。
• 重试机制：网络波动或Facebook API报错时，要做好重试逻辑，重试时需要重新拉取对应范围的S3分片，避免数据不一致。

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