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

相关文章

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

《SQLServer清除日志文件ERRORLOG和删除tempdb.mdf》数据库再使用一段时间后,日志文件会增大,特别是在磁盘容量不足的情况下,更是需要缩减,以下为缩减方法:如果可以停止SQLSe... 目录缩减 ERRORLOG 文件(停止服务后)停止 SQL Server 服务:找到错误日志文件:删除

Windows Server服务器上配置FileZilla后,FTP连接不上?

《WindowsServer服务器上配置FileZilla后,FTP连接不上?》WindowsServer服务器上配置FileZilla后,FTP连接错误和操作超时的问题,应该如何解决?首先,通过... 目录在Windohttp://www.chinasem.cnws防火墙开启的情况下,遇到的错误如下:无法与

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

《一文详解SQLServer如何跟踪自动统计信息更新》SQLServer数据库中,我们都清楚统计信息对于优化器来说非常重要,所以本文就来和大家简单聊一聊SQLServer如何跟踪自动统计信息更新吧... SQL Server数据库中,我们都清楚统计信息对于优化器来说非常重要。一般情况下,我们会开启"自动更新

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

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

JAVA虚拟机中 -D, -X, -XX ,-server参数使用

《JAVA虚拟机中-D,-X,-XX,-server参数使用》本文主要介绍了JAVA虚拟机中-D,-X,-XX,-server参数使用,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有... 目录一、-D参数二、-X参数三、-XX参数总结:在Java开发过程中,对Java虚拟机(JVM)的启动参数进

Windows server服务器使用blat命令行发送邮件

《Windowsserver服务器使用blat命令行发送邮件》在linux平台的命令行下可以使用mail命令来发送邮件,windows平台没有内置的命令,但可以使用开源的blat,其官方主页为ht... 目录下载blatBAT命令行示例备注总结在linux平台的命令行下可以使用mail命令来发送邮件,Win

如何使用Docker部署FTP和Nginx并通过HTTP访问FTP里的文件

《如何使用Docker部署FTP和Nginx并通过HTTP访问FTP里的文件》本文介绍了如何使用Docker部署FTP服务器和Nginx,并通过HTTP访问FTP中的文件,通过将FTP数据目录挂载到N... 目录docker部署FTP和Nginx并通过HTTP访问FTP里的文件1. 部署 FTP 服务器 (

Qt实现发送HTTP请求的示例详解

《Qt实现发送HTTP请求的示例详解》这篇文章主要为大家详细介绍了如何通过Qt实现发送HTTP请求,文中的示例代码讲解详细,具有一定的借鉴价值,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1、添加network模块2、包含改头文件3、创建网络访问管理器4、创建接口5、创建网络请求对象6、创建一个回复对

springMVC返回Http响应的实现

《springMVC返回Http响应的实现》本文主要介绍了在SpringBoot中使用@Controller、@ResponseBody和@RestController注解进行HTTP响应返回的方法,... 目录一、返回页面二、@Controller和@ResponseBody与RestController