Akka（35）： Http：Server side streaming

2024-04-09 04:48
文章标签 http server 35 akka streaming side

本文主要是介绍Akka（35）： Http：Server side streaming，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

   在前面几篇讨论里我们都提到过：Akka-http是一项系统集成工具库。它是以数据交换的形式进行系统集成的。所以，Akka-http的核心功能应该是数据交换的实现了：应该能通过某种公开的数据格式和传输标准比较方便的实现包括异类系统之间通过网上进行的数据交换。覆盖包括：数据编码、发送和数据接收、解析全过程。Akka-http提供了许多网上传输标准数据的概括模型以及数据类型转换方法，可以使编程人员很方便的构建网上往来的Request和Response。但是，现实中的数据交换远远不止针对request和response操作能够满足的。系统之间数据交换经常涉及文件或者数据库表类型的数据上传下载。虽然在Http标准中描述了如何通过MultiPart消息类型进行批量数据的传输，但是这个标准涉及的实现细节包括数据内容描述、数据分段方式、消息数据长度计算等等简直可以立即令人却步。Akka-http是基于Akka-stream开发的：不但它的工作流程可以用Akka-stream来表达，它还支持stream化的数据传输。我们知道：Akka-stream提供了功能强大的FileIO和Data-Streaming，可以用Stream-Source代表文件或数据库数据源。简单来说：Akka-http的消息数据内容HttpEntity可以支持理论上无限长度的data-stream。最可贵的是：这个Source是个Reactive-Stream-Source，具备了back-pressure机制，可以有效应付数据交换参与两方Reactive端点不同的数据传输速率。

  Akka-http的stream类型数据内容是以Source[T,_]类型表示的。首先，Akka-stream通过FileIO对象提供了足够多的file-io操作函数，其中有个fromPath函数可以用某个文件内容数据构建一个Source类型：

/*** Creates a Source from a files contents.* Emitted elements are `chunkSize` sized [[akka.util.ByteString]] elements,* except the final element, which will be up to `chunkSize` in size.** You can configure the default dispatcher for this Source by changing the `akka.stream.blocking-io-dispatcher` or* set it for a given Source by using [[akka.stream.ActorAttributes]].** It materializes a [[Future]] of [[IOResult]] containing the number of bytes read from the source file upon completion,* and a possible exception if IO operation was not completed successfully.** @param f         the file path to read from* @param chunkSize the size of each read operation, defaults to 8192*/def fromPath(f: Path, chunkSize: Int = 8192): Source[ByteString, Future[IOResult]] =fromPath(f, chunkSize, startPosition = 0)

这个函数构建了Source[ByteString,Future[IOResult]]，我们需要把ByteString转化成MessageEntity。首先需要在implicit-scope内提供Marshaller[ByteString,MessageEntity]类型的隐式实例： 

trait JsonCodec extends Json4sSupport {import org.json4s.DefaultFormatsimport org.json4s.ext.JodaTimeSerializersimplicit val serilizer = jackson.Serializationimplicit val formats = DefaultFormats ++ JodaTimeSerializers.all
}
object JsConverters extends JsonCodecobject ServerStreaming extends App {import JsConverters._
...

我们还需要Json-Streaming支持： 

  implicit val jsonStreamingSupport = EntityStreamingSupport.json().withParallelMarshalling(parallelism = 8, unordered = false)

FileIO是blocking操作，我们还可以选用独立的线程供blocking操作使用：

   FileIO.fromPath(file, 256).withAttributes(ActorAttributes.dispatcher("akka.http.blocking-ops-dispatcher"))

现在我们可以从在server上用一个文件构建Source然后再转成Response：

  val route =get {path("files"/Remaining) { name =>complete(loadFile(name))} }def loadFile(path: String) = {//   implicit val ec = httpSys.dispatchers.lookup("akka.http.blocking-ops-dispatcher")val file = Paths.get("/Users/tiger/"+path)FileIO.fromPath(file, 256).withAttributes(ActorAttributes.dispatcher("akka.http.blocking-ops-dispatcher")).map(_.utf8String)}

同样，我们也可以把数据库表内数据转成Akka-Stream-Source，然后再实现到MessageEntity的转换。转换过程包括用Query读取数据库表内数据后转成Reactive-Publisher，然后把publisher转成Akka-Stream-Source，如下： 

object SlickDAO {import slick.jdbc.H2Profile.api._val dbConfig: slick.basic.DatabaseConfig[slick.jdbc.H2Profile] = slick.basic.DatabaseConfig.forConfig("slick.h2")val db = dbConfig.dbcase class CountyModel(id: Int, name: String)case class CountyTable(tag: Tag) extends Table[CountyModel](tag,"COUNTY") {def id = column[Int]("ID",O.AutoInc,O.PrimaryKey)def name = column[String]("NAME",O.Length(64))def * = (id,name)<>(CountyModel.tupled,CountyModel.unapply)}val CountyQuery = TableQuery[CountyTable]def loadTable(filter: String) = {//   implicit val ec = httpSys.dispatchers.lookup("akka.http.blocking-ops-dispatcher")val qry = CountyQuery.filter {_.name.toUpperCase like s"%${filter.toUpperCase}%"}val publisher = db.stream(qry.result)Source.fromPublisher(publisher = publisher).withAttributes(ActorAttributes.dispatcher("akka.http.blocking-ops-dispatcher"))}
}

然后进行到MessageEntity的转换： 

  val route =get {path("files"/Remaining) { name =>complete(loadFile(name))} ~path("tables"/Segment) { t =>complete(SlickDAO.loadTable(t))}}

下面是本次示范的完整源代码： 

import java.nio.file._
import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.common._
import de.heikoseeberger.akkahttpjson4s.Json4sSupport
import org.json4s.jacksonobject SlickDAO {import slick.jdbc.H2Profile.api._val dbConfig: slick.basic.DatabaseConfig[slick.jdbc.H2Profile] = slick.basic.DatabaseConfig.forConfig("slick.h2")val db = dbConfig.dbcase class CountyModel(id: Int, name: String)case class CountyTable(tag: Tag) extends Table[CountyModel](tag,"COUNTY") {def id = column[Int]("ID",O.AutoInc,O.PrimaryKey)def name = column[String]("NAME",O.Length(64))def * = (id,name)<>(CountyModel.tupled,CountyModel.unapply)}val CountyQuery = TableQuery[CountyTable]def loadTable(filter: String) = {//   implicit val ec = httpSys.dispatchers.lookup("akka.http.blocking-ops-dispatcher")val qry = CountyQuery.filter {_.name.toUpperCase like s"%${filter.toUpperCase}%"}val publisher = db.stream(qry.result)Source.fromPublisher(publisher = publisher).withAttributes(ActorAttributes.dispatcher("akka.http.blocking-ops-dispatcher"))}
}trait JsonCodec extends Json4sSupport {import org.json4s.DefaultFormatsimport org.json4s.ext.JodaTimeSerializersimplicit val serilizer = jackson.Serializationimplicit val formats = DefaultFormats ++ JodaTimeSerializers.all
}
object JsConverters extends JsonCodecobject ServerStreaming extends App {import JsConverters._implicit val httpSys = ActorSystem("httpSystem")implicit val httpMat = ActorMaterializer()implicit val httpEC = httpSys.dispatcherimplicit val jsonStreamingSupport = EntityStreamingSupport.json().withParallelMarshalling(parallelism = 8, unordered = false)val (port, host) = (8011,"localhost")val route =get {path("files"/Remaining) { name =>complete(loadFile(name))} ~path("tables"/Segment) { t =>complete(SlickDAO.loadTable(t))}}def loadFile(path: String) = {//   implicit val ec = httpSys.dispatchers.lookup("akka.http.blocking-ops-dispatcher")val file = Paths.get("/Users/tiger/"+path)FileIO.fromPath(file, 256).withAttributes(ActorAttributes.dispatcher("akka.http.blocking-ops-dispatcher")).map(_.utf8String)}val bindingFuture = Http().bindAndHandle(route,host,port)println(s"Server running at $host $port. Press any key to exit ...")scala.io.StdIn.readLine()bindingFuture.flatMap(_.unbind()).onComplete(_ => httpSys.terminate())}

resource/application.conf 

akka {http {blocking-ops-dispatcher {type = Dispatcherexecutor = "thread-pool-executor"thread-pool-executor {// or in Akka 2.4.2+fixed-pool-size = 16}throughput = 100}}
}
slick {h2 {driver = "slick.driver.H2Driver$"db {url = "jdbc:h2:~/slickdemo;mv_store=false"driver = "org.h2.Driver"connectionPool = HikariCPnumThreads = 48maxConnections = 48minConnections = 12keepAliveConnection = true}}
}




这篇关于Akka（35）： Http：Server side streaming的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/887203。

相关文章

SQL server配置管理器找不到如何打开它

SQL server配置管理器找不到如何打开它

《SQLserver配置管理器找不到如何打开它》最近遇到了SQLserver配置管理器打不开的问题,尝试在开始菜单栏搜SQLServerManager无果,于是将自己找到的方法总结分享给大家,对SQ... 目录方法一：桌面图标进入方法二：运行窗口进入方法三：查找文件路径方法四：检查 SQL Server 安
阅读更多...
python连接本地SQL server详细图文教程

python连接本地SQL server详细图文教程

《python连接本地SQLserver详细图文教程》在数据分析领域,经常需要从数据库中获取数据进行分析和处理,下面：本文主要介绍python连接本地SQLserver的相关资料,文中通过代码... 目录一.设置本地账号1.新建用户2.开启双重验证3,开启TCP/IP本地服务二js.python连接实例1.
阅读更多...
Nginx中配置HTTP/2协议的详细指南

Nginx中配置HTTP/2协议的详细指南

《Nginx中配置HTTP/2协议的详细指南》HTTP/2是HTTP协议的下一代版本,旨在提高性能、减少延迟并优化现代网络环境中的通信效率,本文将为大家介绍Nginx配置HTTP/2协议想详细步骤,需... 目录一、HTTP/2 协议概述1.HTTP/22. HTTP/2 的核心特性3. HTTP/2 的优
阅读更多...
使用Python自建轻量级的HTTP调试工具

使用Python自建轻量级的HTTP调试工具

《使用Python自建轻量级的HTTP调试工具》这篇文章主要为大家详细介绍了如何使用Python自建一个轻量级的HTTP调试工具,文中的示例代码讲解详细,感兴趣的小伙伴可以参考一下... 目录一、为什么需要自建工具二、核心功能设计三、技术选型四、分步实现五、进阶优化技巧六、使用示例七、性能对比八、扩展方向建
阅读更多...
使用Python实现快速搭建本地HTTP服务器

使用Python实现快速搭建本地HTTP服务器

《使用Python实现快速搭建本地HTTP服务器》：本文主要介绍如何使用Python快速搭建本地HTTP服务器,轻松实现一键HTTP文件共享,同时结合二维码技术,让访问更简单,感兴趣的小伙伴可以了... 目录1. 概述2. 快速搭建 HTTP 文件共享服务2.1 核心思路2.2 代码实现2.3 代码解读3.
阅读更多...
mysql出现ERROR 2003 (HY000): Can‘t connect to MySQL server on ‘localhost‘ (10061)的解决方法

mysql出现ERROR 2003 (HY000): Can‘t connect to MySQL server on ‘localhost‘ (10061)的解决方法

《mysql出现ERROR2003(HY000):Can‘tconnecttoMySQLserveron‘localhost‘(10061)的解决方法》本文主要介绍了mysql出现... 目录前言：第一步：第二步：第三步：总结：前言：当你想通过命令窗口想打开mysql时候发现提http://www.cpp
阅读更多...
SQL Server清除日志文件ERRORLOG和删除tempdb.mdf

SQL Server清除日志文件ERRORLOG和删除tempdb.mdf

《SQLServer清除日志文件ERRORLOG和删除tempdb.mdf》数据库再使用一段时间后,日志文件会增大,特别是在磁盘容量不足的情况下,更是需要缩减,以下为缩减方法：如果可以停止SQLSe... 目录缩减 ERRORLOG 文件（停止服务后）停止 SQL Server 服务：找到错误日志文件：删除
阅读更多...
Windows Server服务器上配置FileZilla后，FTP连接不上？

Windows Server服务器上配置FileZilla后，FTP连接不上？

《WindowsServer服务器上配置FileZilla后，FTP连接不上？》WindowsServer服务器上配置FileZilla后,FTP连接错误和操作超时的问题,应该如何解决?首先,通过... 目录在Windohttp://www.chinasem.cnws防火墙开启的情况下，遇到的错误如下：无法与
阅读更多...
一文详解SQL Server如何跟踪自动统计信息更新

一文详解SQL Server如何跟踪自动统计信息更新

《一文详解SQLServer如何跟踪自动统计信息更新》SQLServer数据库中,我们都清楚统计信息对于优化器来说非常重要,所以本文就来和大家简单聊一聊SQLServer如何跟踪自动统计信息更新吧... SQL Server数据库中，我们都清楚统计信息对于优化器来说非常重要。一般情况下，我们会开启"自动更新
阅读更多...
Go语言中最便捷的http请求包resty的使用详解

Go语言中最便捷的http请求包resty的使用详解

《Go语言中最便捷的http请求包resty的使用详解》go语言虽然自身就有net/http包,但是说实话用起来没那么好用,resty包是go语言中一个非常受欢迎的http请求处理包,下面我们一起来学... 目录安装一、一个简单的get二、带查询参数三、设置请求头、body四、设置表单数据五、处理响应六、超
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2