PlayScala 2.5.x - 实现完全异步非阻塞的流数据导出

joymufeng

发布于 2018-05-17 15:41:53

1.3K0

文章被收录于专栏：Play & Scala 技术分享Play & Scala 技术分享

介绍

从Play2.5.x开始，Play使用Akka Streams实现流处理，废弃了之前的Enumerator/Iteratee Api。根据官方文档描述，迁移至Akka Streams之后，Play2.5.x的整体性能提升了20%，性能提升相当可观。虽然官方已经更新了JavaStream的文档，但是ScalaStream的文档仍然没有更新，已经提了issue，希望能尽快得到反馈。 ReactiveMongo是一个基于Scala开发的完全异步非阻塞、并且提供流处理功能的MongoDB驱动。该项目目前的流处理功能基于Enumerator/Iteratee实现，Akka Stream的实现放在一个单独的项目开发(RM-AkkaStreams)。结合Play和ReactiveMongo二者的流处理功能，我们可以很方便地实现完全异步非阻塞的报表导出功能。

实现

由于ReactiveMongo暂时还没有提供Akka Streams的流处理实现，所以无法直接通过map/flatMap直接返回一个Stream写回响应：

@Singleton
class TestStreamController @Inject()(val reactiveMongoApi: ReactiveMongoApi, implicit val mat: Materializer) extends Controller {
  def qaCol: JSONCollection = reactiveMongoApi.db.collection[JSONCollection]("qa")
  def exportDataStream = Action { implicit request =>
    val source = Source.actorRef[ByteString](10000, OverflowStrategy.fail).mapMaterializedValue { sourceActor =>
      qaCol
        .find(Json.obj())
        .options(QueryOpts(batchSizeN = 1000))
        .cursor[QA]()
        .foldBulks[Int](0){ (index, list) =>
          sourceActor ! ByteString(list.map(qa => qa.question).mkString("\n") + "\n")
          Cursor.Cont(index + 1)
        }
        .map{ _ =>
          sourceActor ! akka.actor.Status.Success(())
        }
    }

    Ok.chunked(source)
      .withHeaders(
      "Content-Type" -> "application/octet-stream",
      "Content-Disposition" -> (s"attachment;filename=export-" + new DateTime().toString("yyyy-MM-dd") + ".txt"))
  }
}

代码第5行Source.actorRef函数启动一个actor实例sourceActor负责收集报表数据，Source.actorRef的第1个参数bufferSize用于指定缓冲区大小，即Play来不及写回响应的数据暂时放在缓冲区，第2个参数overflowStrategy指定缓冲区溢出后的处理策略。第10行foldBulks方法负责批量从MongoDB数据库读取查询结果，然后以消息形式将数据发送给sourceActor，最后发送一个Status.Success消息表明数据已经发送完毕。数据传递过程如下：

foldBulks(读取查询结果) -> sourceActor(收集查询结果) -> source(生产者) -> Ok.chunked(消费者)

下面是浏览器中看到的效果：

图中499KB/s表示当前的下载速度，997KB表示当前累计的下载大小。

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

scala

PlayScala 2.5.x - 实现完全异步非阻塞的流数据导出

PlayScala 2.5.x - 实现完全异步非阻塞的流数据导出

介绍

实现

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐