scalaakka

Akka实战：HTTP大文件断点上传、下载，秒传

, , , , ,

访问：https://github.com/yangbajing/scala-applications/tree/master/file-upload
获取本文所述完整源码，包括Akka HTTP后端和HTML5实现的前端。

在很多应用里面都会有类似大文件上传的需求，但很多时候我们程序员都会以不支持或不好实现将其推脱掉^_^。

这次因为公司项目需要，另一个组的同事使用Spring实现了一版。他们是采用在前端对大文件分块上传，后端在所有分块上传完成后合并的方式实现了。这个方案有一个弊端，若文件很大，后期对分块做合并时会非常的耗磁盘资源。

本文，我使用Akka HTTP和Akka Stream做为后端服务，可以很优雅的实现大文件的断点续传。原理其实非常的简单，前端计算文件的hash（使用sha256），将hash传到后端查询是否有相同文件已上传，若有将返回已上传文件及文件长度（bytes）。这时候前端就可以知道文件的上传进度，进而判断还需要断点续传的偏移量或者已上传完成（这就是秒传）。

这里有一个设计取舍：客户端对单个文件不做分片，从文件头开始上传。这样的一个好处是可简化服务端的实现，同时也可以优化服务端对文件的存储 （同一个文件将一直使用APPEND的方式写入文件，这样可以更高效的利用磁盘IO。同时，不需要分块合并，若文件很大，生成的大量分块在文件上传完成后再次合并会是一个非常大的资源开销） 。

断点下载

这个怎么说呢？断点下载Akka HTTP原生支持^_^。你只需要使用如下代码：

1
2
3
private def downloadRoute: Route = path("download" / Segment) { hash =>
  getFromFile(FileUtils.getLocalPath(hash).toFile)
}

FileRoute#downloadRoute

FileUtils.getLocalPath(hash) 函数通过对hash值（sha256hex）进行计算和拼接，获取实际文件的本地存储路径再交给Akka HTTP提供的getFromFile指令，剩下的工作就交给Akka。

我们可以通过向Akka HTTP发起HEAD请求来查看支持的HTTP功能，看到在反回的header里有Accept-Ranges: bytes，意思是服务端支持使用字节为单位的范围下载（断点下载功能既基于此实现）。

1
2
3
4
5
6
7
8
9
$ curl --head http://localhost:33333/file/download/7d0559e2f7bf42f0c2becc7fbf91b20ca2e7ec373c941fca21314169de9c7ef4
HTTP/1.1 200 OK
Last-Modified: Fri, 28 Dec 2018 14:12:32 GMT
ETag: "132766a7f528d080"
Accept-Ranges: bytes
Server: akka-http/10.1.6
Date: Sat, 29 Dec 2018 02:17:41 GMT
Content-Type: application/octet-stream
Content-Length: 65463496

断点上传

很遗憾，Akka HTTP默认不支持断点上传，这需要自行实现。但是，Akka HTTP做为一个toolkit，足够灵活且强大，实现断点上传功能so easy。

断点上传实现

基于常规的代码设计方式，我们需要Controller、Service，那就先从Controller开始：

FileRoute#uploadRoute

1
2
3
4
5
6
7
8
9
10
11
12
def uploadRoute: Route = path("upload") {
  post {
    withoutSizeLimit {
      entity(as[Multipart.FormData]) { formData =>
        onSuccess(fileService.handleUpload(formData)) { results =>
          import me.yangbajing.fileupload.util.JacksonSupport._
          complete(results)
        }
      }
    }
  }
}

FileRoute#uploadRoute

对于不熟悉Akka HTTP的朋友，推荐阅读我写的电子书（打个广告）：Scala Web 开发。这里需要注意的一个指令是withoutSizeLimit，默认Akka HTTP对请求大小限制比较低，我们可以通过withoutSizeLimit指令取消对单个API的请求大小限制，同时又不影响整个Web服务的大小限制。另外，这里通过entity(as[Multipart.FormData])以Unmarshaller的方式获取整个Multipart.FormData对象并传入 FileService#handleUpload 函数进行处理。

FileService#handleUpload

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
override def handleUpload(formData: Multipart.FormData): Future[Seq[FileBO]] = {
  formData.parts
    .map { part =>
      val (hash, contentLength, startPosition) = part.name.split('.') match {
        case Array(a, b, c) => (a, b.toLong, c.toLong)
        case Array(a, b)    => (a, b.toLong, 0L)
        case Array(a)       => (a, 0L, 0L)
        case _              => throw new IllegalArgumentException(s"Multipart.FormData name格式不符合要求：${part.name}")
      }
      FileInfo(part, hash, contentLength, startPosition)
    }
    .log("fileInfo", info => logger.debug(s"fileInfo: $info"))
    .mapAsync(Constants.FILE_PART_MAX)(processFile)
    .runWith(Sink.seq)
}

formData.parts是一个Akka Stream流，类型签名为Source[Multipart.FormData.BodyPart, Any]。有关Akka Stream更详细的资料请参阅Akka Streams官方文档。这里，每个part都代表FormData的一个字段（对应HTML 5的FormData类型，同时前端需要使用Content-Type: multipart/form-data; boundary=----WebKitFormBoundaryrP4DAyu31ilqEWmz方式发起请求）。每个part.name都是用英号逗号分隔的三部分来做为请求的字段名，分别是：<hash>.<content length>.<start position>，这样我们就可以在不加入任何其它字段的情况下告知服务端当前上传文件的sha256hex计算出的hash值、文件大小（bytes）和上传起始偏移量。

FileUtils#uploadFile

文件上传的核心逻辑在 FileUtils#uploadFile 函数：

1
2
3
4
5
def uploadFile(fileInfo: FileInfo)(implicit mat: Materializer, ec: ExecutionContext): Future[FileBO] = {
  val maybeMeta = FileUtils.getFileMeta(fileInfo.hash)
  val beContinue = maybeMeta.isDefined && fileInfo.startPosition > 0L
  if (beContinue) uploadContinue(fileInfo, maybeMeta.get) else uploadNewFile(fileInfo)
}

uploadFile函数根据是否为续传来分别调用uploadContinue或uploadNewFile函数。首先来看看新文件上传时的代码逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def uploadNewFile(fileInfo: FileInfo)(implicit mat: Materializer, ec: ExecutionContext) = {
  val bodyPart = fileInfo.bodyPart
  val tmpPath =
    if (fileInfo.validHash) FileUtils.getLocalPath(fileInfo.hash)
    else Files.createTempFile(FileUtils.TMP_DIR, bodyPart.filename.getOrElse(""), "") // (1)
  val sha = MessageDigest.getInstance("SHA-256")
  bodyPart.entity.dataBytes
    .map { byteString =>
      byteString.asByteBuffers.foreach(sha.update) // (2)
      byteString
    }
    .runWith(FileIO.toPath(tmpPath)) // (3)
    .map { ioResult =>
      val hash = Utils.bytesToHex(sha.digest()) // (4)
      if (fileInfo.validHash) {
        require(fileInfo.hash == hash, s"前端上传hash与服务端计算hash值不匹配，${fileInfo.hash} != $hash")
      }
      val localPath = if (fileInfo.validHash) tmpPath else move(hash, tmpPath, ioResult.count) // (5)
      FileBO(hash, localPath, ioResult.count, bodyPart.filename, bodyPart.headers)
    }
}
  1. 根据前端是否上传了有效的hash值（sha256hex）来判断是把文件先写入临时文件还是直接写入实际的本地存储位置（根据hash值计算出本地实际的存储位置）。
  2. Akka HTTP中，上传的文件以流的方式进入，在此对每个ByteString计算并更新sha256值。
  3. 在Akka Stream的Sink端，接收流传入的元素并写入本地文件。
  4. 文件写入结束后调用sha.digest()方法获取已上传文件的sha256值。
  5. 根据是否临时文件来判断是否需要将临时文件移动到实际的本地存储路径，通过文件的hash值来计算出实际的本地存储路径。
1
2
3
4
5
6
7
8
9
def uploadContinue(fileInfo: FileInfo, meta: FileMeta)(implicit mat: Materializer, ec: ExecutionContext) = {
  val bodyPart = fileInfo.bodyPart
  val localPath = FileUtils.getLocalPath(fileInfo.hash)
  logger.debug(s"断点续传，startPosition：${fileInfo.startPosition}，路径：$localPath")
  val hash = fileInfo.hash
  bodyPart.entity.dataBytes
    .runWith(FileIO.toPath(localPath, Set(APPEND), fileInfo.startPosition)) // (1)
    .map(ioResult => FileBO(hash, localPath, meta.size + ioResult.count, bodyPart.filename, bodyPart.headers))
}

断点上传时的逻辑其实相对简单，需要注意的是在(1)处调用FileIO.toPath将流定入本地时需要以APPEND模式追加写入到已存在文件。

秒传

在已实现断点上传功能之上，秒传的实现逻辑就非常清晰了。客户端在调用file/upload接口上传文件之前先调用/file/progress/{hash}接口判断相同hash值文件的上传情况，再决定下一步处理。

  1. 客户端计算文件hash，并以文件hash和文件大小作为参数调用/file/progress/{hash}接口
  2. 服务端根据上传hash值判断文件是否已上传，若存在返回已上传文档大小（bytes）
  3. 客户端收到服务端响应后根据文件是否存在及已存在文件大小判断秒传、断点续传还是新上传
  4. 秒传，返回文件长度与当前准备上传文件长度大小一致
  5. 断点续传，返回文件大小比当前准备上传文件长度小
  6. 新上传，返回文件不存在
  7. 其它情况，作为新文件上传

秒传的代码逻辑台下：

1
2
3
4
5
6
7
8
def progressRoute: Route = path("progress" / Segment) { hash =>
  onSuccess(fileService.progressByHash(hash)) {
    case Some(v) =>
      import me.yangbajing.fileupload.util.JacksonSupport._
      complete(v)
    case None => complete(StatusCodes.NotFound)
  }
}

文件上传进度接口定义如上。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// FileServiceImpl.scala
override def progressByHash(hash: String): Future[Option[FileMeta]] = {
  require(Objects.nonNull(hash) && hash.nonEmpty, "hash 不能为空。")
  Future.successful(FileUtils.getFileMeta(hash))
}

// FileUtils.getFileMeta
def getFileMeta(hash: String): Option[FileMeta] = {
  if (hash == null || Constants.HASH_LENGTH != hash.length) {
    None
  } else {
    val path = getLocalPath(hash)
    if (Files.exists(path) && Files.isReadable(path)) Some(FileMeta(hash, Files.size(path), path)) else None
  }
}

文件上传进度服务实现定义如上。

小结

本文以Akka HTTP和HTML 5讲述了怎样实现一个支持大文件断点上传、下载和秒传的示例应用程序。

分享到