文档章节

Akka实战:HTTP大文件断点上传、下载,秒传

羊八井
 羊八井
发布于 02/23 20:39
字数 2011
阅读 17
收藏 0

访问: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原生支持^_^。你只需要使用如下代码:

  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,意思是服务端支持使用字节为单位的范围下载(断点下载功能既基于此实现)。

$ 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。

断点上传实现

基于常规的代码设计方式,我们需要ControllerService,那就先从Controller开始:

FileRoute#uploadRoute

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

  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 函数:

  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函数根据是否为续传来分别调用uploadContinueuploadNewFile函数。首先来看看新文件上传时的代码逻辑:

  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值来计算出实际的本地存储路径。
  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. 其它情况,作为新文件上传

秒传的代码逻辑台下:

  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)
    }
  }

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

  // 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讲述了怎样实现一个支持大文件断点上传、下载和秒传的示例应用程序。

© 著作权归作者所有

共有 人打赏支持
羊八井

羊八井

粉丝 93
博文 40
码字总数 50313
作品 0
渝北
技术主管
私信 提问
go-fastdfs v1.2.2 发布,高性能、高可靠分布式文件系统

go-fastdfs是一个基于http协议的分布式文件系统,它基于大道至简的设计理念,一切从简设计,使得它的运维及扩展变得更加简单,它具有高性能、高可靠、无中心、免维护等优点。 大家使用前一定...

s_jqzhang
03/17
0
0
高性能、高可靠分布式文件系统 go-fastdfs v1.2.1 发布

go-fastdfs是一个基于http协议的分布式文件系统,它基于大道至简的设计理念,一切从简设计,使得它的运维及扩展变得更加简单,它具有高性能、高可靠、无中心、免维护等优点。 大家使用前一定...

s_jqzhang
03/12
0
0
高性能、高可靠分布式文件系统 go-fastdfs v1.2.0 发布

go-fastdfs是一个基于http协议的分布式文件系统,它基于大道至简的设计理念,一切从简设计,使得它的运维及扩展变得更加简单,它具有高性能、高可靠、无中心、免维护等优点。 优点 支持curl命...

s_jqzhang
03/08
0
0
网盘--iBarn

iBarn是一个先进的网盘系统,提供文件的网络备份,同步和分享服务。支持断点续传,秒传等功能。 可选择文件下载到本地或者在线收藏;回收站功能防止用户误删数据。云存储的不二之选 功能列表...

飘渺孤鸿影love
2015/09/10
6.4K
2
令狐大侠/大文件上传断点续传fastdfs

大文件上传,断点续传,秒传,fastdfs 项目介绍 实现h5与fastdfs之间的断点续传,大文件上传,秒传 软件架构 软件架构说明 webuploader+springboot+redis+fastdfs(服务端)+FastDFS_Client(非官网,...

令狐大侠
2018/06/19
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Perl - 获取文件项目

参考:http://www.runoob.com/perl/perl-directories.html 下面返回JSON格式的文件列表 #!/usr/bin/perluse strict;use warnings;use utf8;use feature ':5.26';require Fi......

wffger
昨天
1
0
vue组件系列3、查询下载

直接源码,虽然样式样式不好看,逻辑也不是最优,但是可以留作纪念。毕竟以后类似的功能只需要优化就可以了,不用每次都重头开始。。。 <template> <div class="pre_upload"> <div ...

轻轻的往前走
昨天
1
0
java浅复制和深复制

之前写了数组的复制,所以这里继续总结一下浅复制和深复制。 浅拷贝:对基本数据类型进行值传递,对引用数据类型进行引用传递般的拷贝。 深拷贝:对基本数据类型进行值传递,对引用数据类型,...

woshixin
昨天
1
0
kubernetes 二进制包安装

环境 角色 主机名 内网 IP 集群 IP 操作系统 服务 执行目录 部署机 k8s-master master120 10.0.4.120 - CentOS kube-apiserver kube-scheduler kube-controller-manager /opt/kubernetes/ et......

Colben
昨天
8
0
瑞达利欧的《原则》读后感作文2400字

瑞达利欧的《原则》读后感作文2400字: 拿到同事送的书读了几十页后,我意识到自己犯了一个错误:之前以为像《原则》这个书名,配上最成功的对冲基金经理、受用终身的教诲、傅盛引用并盛赞,...

原创小博客
昨天
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部