Akka（41）： Http：DBTable-rows streaming - 数据库表行交换

2024-04-09 04:48
文章标签 http 数据库 41 交换 rows akka streaming dbtable 表行

本文主要是介绍Akka（41）： Http：DBTable-rows streaming - 数据库表行交换，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

  在前面一篇讨论里我们介绍了通过http进行文件的交换。因为文件内容是以一堆bytes来表示的，而http消息的数据部分也是byte类型的，所以我们可以直接用Source[ByteString,_]来读取文件然后放进HttpEntity中。我们还提到：如果需要进行数据库数据交换的话，可以用Source[ROW,_]来表示库表行，但首先必须进行ROW -> ByteString的转换。在上期讨论我们提到过这种转换其实是ROW->Json->ByteString或者反方向的转换，在Akka-http里称之为Marshalling和Unmarshalling。Akka-http的Marshalling实现采用了type-class编程模式，需要为每一种类型与Json的转换在可视域内提供Marshaller[A,B]类型的隐式实例。Akka-http默认的Json工具库是Spray-Json，着重case class，而且要提供JsonFormat?(case-class),其中?代表case class的参数个数，用起来略显复杂。不过因为是Akka-http的配套库，在将来Akka-http的持续发展中具有一定的优势，所以我们还是用它来进行下面的示范。

下面就让我们开始写些代码吧。首先，我们用一个case class代表数据库表行结构，然后用它作为流元素来构建一个Source，如下：

  case class County(id: Int, name: String)val source: Source[County, NotUsed] = Source(1 to 5).map { i => County(i, s"中国广东省地区编号 #$i") }

我们先设计服务端的数据下载部分： 

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives._
import akka._
import akka.http.scaladsl.common._
import spray.json.DefaultJsonProtocol
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupporttrait MyFormats extends SprayJsonSupport with DefaultJsonProtocol
object Converters extends MyFormats {case class County(id: Int, name: String)val source: Source[County, NotUsed] = Source(1 to 5).map { i => County(i, s"中国广东省地区编号 #$i") }implicit val countyFormat = jsonFormat2(County)
}object HttpDBServer extends App {import Converters._implicit val httpSys = ActorSystem("httpSystem")implicit val httpMat = ActorMaterializer()implicit val httpEC = httpSys.dispatcherimplicit val jsonStreamingSupport = EntityStreamingSupport.json().withParallelMarshalling(parallelism = 8, unordered = false)val route =path("rows") {get {complete {source}}}val (port, host) = (8011,"localhost")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())}

在上面的代码里我们直接把source放进了complete()，然后期望这个directive能通过ToEntityMarshaller[County]类实例用Spray-Json把Source[County,NotUsed]转换成Source[ByteString,NotUsed]然后放入HttpResponse的HttpEntity里。转换结果只能在客户端得到证实。我们知道HttpResponse里的Entity.dataBytes就是一个Source[ByteString,_]，我们可以把它Unmarshall成Source[County,_]，然后用Akka-stream来操作：

      case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>val futSource = Unmarshal(entity).to[Source[County,NotUsed]]futSource.onSuccess {case source => source.runForeach(println)}


上面这个Unmarshal调用了下面这个FromEntityUnmarshaller[County]隐式实例:

  // support for as[Source[T, NotUsed]]implicit def sprayJsonSourceReader[T](implicit reader: RootJsonReader[T], support: EntityStreamingSupport): FromEntityUnmarshaller[Source[T, NotUsed]] =Unmarshaller.withMaterializer { implicit ec ⇒ implicit mat ⇒ e ⇒if (support.supported.matches(e.contentType)) {val frames = e.dataBytes.via(support.framingDecoder)val unmarshal = sprayJsonByteStringUnmarshaller(reader)(_)val unmarshallingFlow =if (support.unordered) Flow[ByteString].mapAsyncUnordered(support.parallelism)(unmarshal)else Flow[ByteString].mapAsync(support.parallelism)(unmarshal)val elements = frames.viaMat(unmarshallingFlow)(Keep.right)FastFuture.successful(elements)} else FastFuture.failed(Unmarshaller.UnsupportedContentTypeException(support.supported))}

这个隐式实例是由Spray-Jason提供的，在SprayJsonSupport.scala里。
 下面是这部分客户端的完整代码： 

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import scala.util._
import akka._
import akka.http.scaladsl.common._
import spray.json.DefaultJsonProtocol
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.unmarshalling.Unmarshaltrait MyFormats extends SprayJsonSupport with DefaultJsonProtocol
object Converters extends MyFormats {case class County(id: Int, name: String)implicit val countyFormat = jsonFormat2(County)
}object HttpDBClient extends App {import Converters._implicit val sys = ActorSystem("ClientSys")implicit val mat = ActorMaterializer()implicit val ec = sys.dispatcherimplicit val jsonStreamingSupport: JsonEntityStreamingSupport = EntityStreamingSupport.json()def downloadRows(request: HttpRequest) = {val futResp = Http(sys).singleRequest(request)futResp.andThen {case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>val futSource = Unmarshal(entity).to[Source[County,NotUsed]]futSource.onSuccess {case source => source.runForeach(println)}case Success(r@HttpResponse(code, _, _, _)) =>println(s"download request failed, response code: $code")r.discardEntityBytes()case Success(_) => println("Unable to download rows!")case Failure(err) => println(s"download failed: ${err.getMessage}")}}downloadRows(HttpRequest(HttpMethods.GET,uri = s"http://localhost:8011/rows"))scala.io.StdIn.readLine()sys.terminate()}

以上我们已经实现了客户端从服务端下载一段数据库表行，然后以Akka-stream的操作方式来处理下载数据。那么反向交换即从客户端上传一段表行的话就需要把一个Source[T,_]转换成Source[ByteString,_]然后放进HttpRequest的HttpEntity里。服务端收到数据后又要进行反向的转换即把Request.Entity.dataBytes从Source[ByteString,_]转回Source[T,_]。Akka-http在客户端没有提供像complete这样的强大的自动化功能。我们可能需要自定义并提供像ToRequestMarshaller[Source[T,_]]这样的隐式实例。但Akka-http的Marshalling-type-class是个非常复杂的系统。如果我们的目的是简单提供一个Source[ByteString,_]，我们是否可以直接调用Spray-Json的函数来进行ROW->Son->ByteString转换呢？如下：

  import akka.util.ByteStringimport akka.http.scaladsl.model.HttpEntity.limitableByteSourceval source: Source[County,NotUsed] = Source(1 to 5).map {i => County(i, s"广西壮族自治区地市县编号 #$i")}def countyToByteString(c: County) = {ByteString(c.toJson.toString)}val flowCountyToByteString : Flow[County,ByteString,NotUsed] = Flow.fromFunction(countyToByteString)val rowBytes = limitableByteSource(source via flowCountyToByteString)val request = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/rows")val data = HttpEntity(ContentTypes.`application/json`,rowBytes)

我们直接用toJson函数进行County->Json转换实现了flowCountyToByteString。toJason是Spray-Json提供的一个函数： 

package json {case class DeserializationException(msg: String, cause: Throwable = null, fieldNames: List[String] = Nil) extends RuntimeException(msg, cause)class SerializationException(msg: String) extends RuntimeException(msg)private[json] class PimpedAny[T](any: T) {def toJson(implicit writer: JsonWriter[T]): JsValue = writer.write(any)}private[json] class PimpedString(string: String) {@deprecated("deprecated in favor of parseJson", "1.2.6")def asJson: JsValue = parseJsondef parseJson: JsValue = JsonParser(string)}
}

假设服务端收到数据后以Akka-stream方式再转换成一个List返回，我们用下面的方法来测试功能： 

  def uploadRows(request: HttpRequest, dataEntity: RequestEntity) = {val futResp = Http(sys).singleRequest(request.copy(entity = dataEntity))futResp.andThen {case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>entity.dataBytes.map(_.utf8String).runForeach(println)case Success(r@HttpResponse(code, _, _, _)) =>println(s"Upload request failed, response code: $code")r.discardEntityBytes()case Success(_) => println("Unable to Upload file!")case Failure(err) => println(s"Upload failed: ${err.getMessage}")}}

服务端接收数据处理方法如下： 

     post {withoutSizeLimit {entity(asSourceOf[County]) { source =>val futofNames: Future[List[String]] =source.runFold(List[String](""))((acc, b) => acc ++ List(b.name))complete {futofNames}}}}

考虑到在数据转换的过程中可能会出现异常。需要异常处理方法来释放backpressure： 

  def postExceptionHandler: ExceptionHandler =ExceptionHandler {case _: RuntimeException =>extractRequest { req =>req.discardEntityBytes()complete((StatusCodes.InternalServerError.intValue,"Upload Failed!"))}}post {withoutSizeLimit {handleExceptions(postExceptionHandler) {entity(asSourceOf[County]) { source =>val futofNames: Future[List[String]] =source.runFold(List[String](""))((acc, b) => acc ++ List(b.name))complete {futofNames}}}}}

在客户端试运行返回结果显示： 

  uploadRows(request,data)["","广西壮族自治区地市县编号 #1","广西壮族自治区地市县编号 #2","广西壮族自治区地市县编号 #3","广西壮族自治区地市县编号 #4","广西壮族自治区地市县编号 #5"]

正是我们期待的结果。

下面是本次讨论的示范代码：

服务端：

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka._
import akka.http.scaladsl.common._
import spray.json.DefaultJsonProtocol
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import scala.concurrent._
import akka.http.scaladsl.server._
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.model._trait MyFormats extends SprayJsonSupport with DefaultJsonProtocol
object Converters extends MyFormats {case class County(id: Int, name: String)val source: Source[County, NotUsed] = Source(1 to 5).map { i => County(i, s"中国广东省地区编号 #$i") }implicit val countyFormat = jsonFormat2(County)
}object HttpDBServer extends App {import Converters._implicit val httpSys = ActorSystem("httpSystem")implicit val httpMat = ActorMaterializer()implicit val httpEC = httpSys.dispatcherimplicit val jsonStreamingSupport = EntityStreamingSupport.json().withParallelMarshalling(parallelism = 8, unordered = false)def postExceptionHandler: ExceptionHandler =ExceptionHandler {case _: RuntimeException =>extractRequest { req =>req.discardEntityBytes()complete((StatusCodes.InternalServerError.intValue,"Upload Failed!"))}}val route =path("rows") {get {complete {source}} ~post {withoutSizeLimit {handleExceptions(postExceptionHandler) {entity(asSourceOf[County]) { source =>val futofNames: Future[List[String]] =source.runFold(List[String](""))((acc, b) => acc ++ List(b.name))complete {futofNames}}}}}}val (port, host) = (8011,"localhost")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())}

客户端： 

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import scala.util._
import akka._
import akka.http.scaladsl.common._
import spray.json.DefaultJsonProtocol
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.unmarshalling._trait MyFormats extends SprayJsonSupport with DefaultJsonProtocol
object Converters extends MyFormats {case class County(id: Int, name: String)implicit val countyFormat = jsonFormat2(County)
}object HttpDBClient extends App {import Converters._implicit val sys = ActorSystem("ClientSys")implicit val mat = ActorMaterializer()implicit val ec = sys.dispatcherimplicit val jsonStreamingSupport: JsonEntityStreamingSupport = EntityStreamingSupport.json()def downloadRows(request: HttpRequest) = {val futResp = Http(sys).singleRequest(request)futResp.andThen {case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>val futSource = Unmarshal(entity).to[Source[County,NotUsed]]futSource.onSuccess {case source => source.runForeach(println)}case Success(r@HttpResponse(code, _, _, _)) =>println(s"download request failed, response code: $code")r.discardEntityBytes()case Success(_) => println("Unable to download rows!")case Failure(err) => println(s"download failed: ${err.getMessage}")}}downloadRows(HttpRequest(HttpMethods.GET,uri = s"http://localhost:8011/rows"))import akka.util.ByteStringimport akka.http.scaladsl.model.HttpEntity.limitableByteSourceval source: Source[County,NotUsed] = Source(1 to 5).map {i => County(i, s"广西壮族自治区地市县编号 #$i")}def countyToByteString(c: County) = {ByteString(c.toJson.toString)}val flowCountyToByteString : Flow[County,ByteString,NotUsed] = Flow.fromFunction(countyToByteString)val rowBytes = limitableByteSource(source via flowCountyToByteString)val request = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/rows")val data = HttpEntity(ContentTypes.`application/json`,rowBytes)def uploadRows(request: HttpRequest, dataEntity: RequestEntity) = {val futResp = Http(sys).singleRequest(request.copy(entity = dataEntity))futResp.andThen {case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>entity.dataBytes.map(_.utf8String).runForeach(println)case Success(r@HttpResponse(code, _, _, _)) =>println(s"Upload request failed, response code: $code")r.discardEntityBytes()case Success(_) => println("Unable to Upload file!")case Failure(err) => println(s"Upload failed: ${err.getMessage}")}}uploadRows(request,data)scala.io.StdIn.readLine()sys.terminate()}





这篇关于Akka（41）： Http：DBTable-rows streaming - 数据库表行交换的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

C# WinForms存储过程操作数据库的实例讲解

C# WinForms存储过程操作数据库的实例讲解

《C#WinForms存储过程操作数据库的实例讲解》：本文主要介绍C#WinForms存储过程操作数据库的实例,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、存储过程基础二、C# 调用流程1. 数据库连接配置2. 执行存储过程（增删改）3. 查询数据三、事务处
阅读更多...
mysql数据库重置表主键id的实现

mysql数据库重置表主键id的实现

《mysql数据库重置表主键id的实现》在我们的开发过程中,难免在做测试的时候会生成一些杂乱无章的SQL主键数据,本文主要介绍了mysql数据库重置表主键id的实现,具有一定的参考价值,感兴趣的可以了... 目录关键语法演示案例在我们的开发过程中，难免在做测试的时候会生成一些杂乱无章的SQL主键数据，当我们
阅读更多...
Spring Boot 整合 MyBatis 连接数据库及常见问题

Spring Boot 整合 MyBatis 连接数据库及常见问题

《SpringBoot整合MyBatis连接数据库及常见问题》MyBatis是一个优秀的持久层框架,支持定制化SQL、存储过程以及高级映射,下面详细介绍如何在SpringBoot项目中整合My... 目录一、基本配置1. 添加依赖2. 配置数据库连接二、项目结构三、核心组件实现（示例）1. 实体类2. Ma
阅读更多...
C语言实现两个变量值交换的三种方式

C语言实现两个变量值交换的三种方式

《C语言实现两个变量值交换的三种方式》两个变量值的交换是编程中最常见的问题之一,以下将介绍三种变量的交换方式,其中第一种方式是最常用也是最实用的,后两种方式一般只在特殊限制下使用,需要的朋友可以参考下... 目录1.使用临时变量（推荐）2.相加和相减的方式（值较大时可能丢失数据）3.按位异或运算1.使用临时
阅读更多...
使用C语言实现交换整数的奇数位和偶数位

使用C语言实现交换整数的奇数位和偶数位

《使用C语言实现交换整数的奇数位和偶数位》在C语言中,要交换一个整数的二进制位中的奇数位和偶数位,重点需要理解位操作,当我们谈论二进制位的奇数位和偶数位时,我们是指从右到左数的位置,本文给大家介绍了使... 目录一、问题描述二、解决思路三、函数实现四、宏实现五、总结一、问题描述使用C语言代码实现：将一个整
阅读更多...
查看Oracle数据库中UNDO表空间的使用情况(最新推荐)

查看Oracle数据库中UNDO表空间的使用情况(最新推荐)

《查看Oracle数据库中UNDO表空间的使用情况(最新推荐)》Oracle数据库中查看UNDO表空间使用情况的4种方法：DBA_TABLESPACES和DBA_DATA_FILES提供基本信息,V$... 目录1. 通过 DBjavascriptA_TABLESPACES 和 DBA_DATA_FILES
阅读更多...
Java实现数据库图片上传与存储功能

Java实现数据库图片上传与存储功能

《Java实现数据库图片上传与存储功能》在现代的Web开发中,上传图片并将其存储在数据库中是常见的需求之一,本文将介绍如何通过Java实现图片上传,存储到数据库的完整过程,希望对大家有所帮助... 目录1. 项目结构2. 数据库表设计3. 实现图片上传功能3.1 文件上传控制器3.2 图片上传服务4. 实现
阅读更多...
Go语言中最便捷的http请求包resty的使用详解

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

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

使用Dify访问mysql数据库详细代码示例

《使用Dify访问mysql数据库详细代码示例》：本文主要介绍使用Dify访问mysql数据库的相关资料,并详细讲解了如何在本地搭建数据库访问服务,使用ngrok暴露到公网,并创建知识库、数据库访... 1、在本地搭建数据库访问的服务，并使用ngrok暴露到公网。#sql_tools.pyfrom
阅读更多...
Java实现数据库图片上传功能详解

Java实现数据库图片上传功能详解

《Java实现数据库图片上传功能详解》这篇文章主要为大家详细介绍了如何使用Java实现数据库图片上传功能,包含从数据库拿图片传递前端渲染,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1、前言2、数据库搭建&nbsChina编程p; 3、后端实现将图片存储进数据库4、后端实现从数据库取出图片给前端5、前端拿到
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2