Akka-CQRS（9）- gRPC，实现前端设备与平台系统的高效集成

前面我们完成了一个CQRS模式的数据采集（录入）平台。可以预见：数据的产生是在线下各式各样的终端系统中，包括web、桌面、移动终端。那么，为了实现一个完整的系统，必须把前端设备通过某种网络连接形式与数据采集平台集成为一体。有两种方式可以实现需要的网络连接：Restful-api, gRPC。由于gRPC支持http/2通讯协议，支持持久连接方式及双向数据流。所以对于POS设备这样的前端选择gRPC作为网络连接方式来实现实时的操作控制应该是正确的选择，毕竟采用恒久连接和双向数据流效率会高很多。gRPC是google公司的标准，基于protobuffer消息：一种二进制序列化数据交换机制。gRPC的优势在这里就不再细说，读者可以参考前面有关gRPC的讨论博文。

下面是系统结构示意图：

这篇讨论焦点集中在gRPC的server,client两头的具体实现。刚才提过，gRPC是google公司的开源库，同时还提供了各种语言的客户端，有：java, C++,python,go ... 但就是没有scala的，只能找第三方的scala客户端了。现在市面可供选择的gRPC-scala-客户端有scalaPB和akka-grpc两个，akka-grpc是基于akka-stream和akka-http构建的，按理来说会更合适，但由于还是处于preview版本，以后再说吧，现在只有scalaPB可选了。scalaPB是一个比较成熟的gRPC客户端，在前面的博客里我们也进行了介绍和示范。下面我们就用scalaPB来实现上面这个例子的客户端-平台集成。

首先，gRPC是通过protobuffer进行序列化数据传输的。下面是这个例子的.proto定义文件：

syntax = "proto3";import "google/protobuf/wrappers.proto";
import "google/protobuf/any.proto";
import "scalapb/scalapb.proto";option (scalapb.options) = {// use a custom Scala package name// package_name: "io.ontherocks.introgrpc.demo"// don't append file name to packageflat_package: true// generate one Scala file for all messages (services still get their own file)single_file: true// add imports to generated file// useful when extending traits or using custom types// import: "io.ontherocks.hellogrpc.RockingMessage"// code to put at the top of generated file// works only with `single_file: true`//preamble: "sealed trait SomeSealedTrait"
};package com.datatech.pos.messages;message PBVchState {      //单据状态string opr  = 1;    //收款员int64  jseq = 2;    //begin journal sequence for read-side replayint32  num  = 3;    //当前单号int32  seq  = 4;    //当前序号bool   void = 5;    //取消模式bool   refd = 6;    //退款模式bool   susp = 7;    //挂单bool   canc = 8;    //废单bool   due  = 9;    //当前余额string su   = 10;   //主管编号string mbr  = 11;   //会员号int32  mode = 12;   //当前操作流程：0=logOff, 1=LogOn, 2=Payment
}message PBTxnItem {       //交易记录string txndate    = 1;   //交易日期string txntime    = 2;   //录入时间string opr        = 3;   //操作员int32  num        = 4;   //销售单号int32  seq        = 5;   //交易序号int32  txntype    = 6;   //交易类型int32  salestype  = 7;   //销售类型int32  qty        = 8;   //交易数量int32  price      = 9;   //单价（分）int32  amount     = 10;  //码洋（分）int32  disc       = 11;  //折扣率 (%)int32  dscamt     = 12;  //折扣额：负值  net实洋 = amount + dscamtstring member     = 13;  //会员卡号string code       = 14;  //编号（商品、卡号...）string acct       = 15;  //账号string dpt        = 16;  //部类
}message PBPOSResponse {int32  sts                  = 1;string msg                  = 2;PBVchState voucher          = 3;repeated PBTxnItem txnitems   = 4;}message PBPOSCommand {int64  shopid = 1;string commandname = 2;string delimitedparams = 3;   //for multiple parameters, use ; to delimit
}service SendCommand {rpc SingleResponse(PBPOSCommand) returns (PBPOSResponse) {};rpc MultiResponse(PBPOSCommand) returns (stream PBPOSResponse) {};
}

前端通过两种模式向平台发送指令PBPOSCommand: SingleResponse代表传统的request/response交互模式，MultiResponse,又或者server-streaming,代表前端发送一个指令，服务端返回一串Response, 或response-stream。Command和PBCommand、POSResponse和PBPOSResponse之间必须具备相互转换函数：

package com.datatech.pos.cloud
import Messages._
import com.datatech.pos.messages._object PBConverter {implicit class PBConverter(pbmsg: PBPOSCommand) {def toPOSComand: POSMessage = pbmsg.commandname.toUpperCase match {case "LOGON" => POSMessage(pbmsg.shopid,LogOn(pbmsg.delimitedparams))case "LOGOFF" => POSMessage(pbmsg.shopid,LogOff)...}}implicit class POSResponseConvert(resp: POSResponse) {def toPBPOSResponse: PBPOSResponse = new PBPOSResponse(sts = resp.sts,msg = resp.msg,voucher = Some(resp.voucher.toPBVchState),txnitems = resp.txnItems.map(_.toPBTxnItem))}implicit class VchStateConvert(state: VchStates) {def toPBVchState: PBVchState = new PBVchState(opr  = state.opr,   //收款员jseq = state.jseq,   //begin journal sequence for read-side replaynum  = state.num,  //当前单号seq  = state.seq,   //当前序号void = state.void,  //取消模式refd = state.refd, //退款模式susp = state.susp,   //挂单canc = state.canc,  //废单due  = state.due,   //当前余额su   = state.su,  //主管编号mbr  = state.mbr,   //会员号mode = state.mode //当前操作流程：0=logOff, 1=LogOn, 2=Payment)}implicit class TxnItemConvert(item: TxnItem) {def toPBTxnItem: PBTxnItem = new PBTxnItem(txndate    = item.txndate,   //交易日期txntime    = item.txntime,   //录入时间opr        = item.opr,   //操作员num        = item.num,   //销售单号seq        = item.seq,  //交易序号txntype    = item.txntype,   //交易类型salestype  = item.salestype,  //销售类型qty        = item.qty,  //交易数量price      = item.price,  //单价（分）amount     = item.amount,  //码洋（分）disc       = item.disc,  //折扣率 (%)dscamt     = item.dscamt, //折扣额：负值  net实洋 = amount + dscamtmember     = item.member,  //会员卡号code       = item.code,  //编号（商品、卡号...）acct       = item.acct,  //账号dpt        = item.dpt  //部类)}
}

然后可以开始实现平台端POS接口服务了：

package com.datatech.pos.cloud
import com.datatech.pos.messages._
import io.grpc.stub.StreamObserver
import PBConverter._
import akka.actor.ActorRef
import akka.pattern.ask
import scala.concurrent.duration._
import akka.util.Timeout
import Messages._
import scala.concurrent.{Await, Future}
import com.typesafe.config.ConfigFactory
import com.datatech.sdp
import sdp.logging._class gRPCServices(writerRouter: ActorRef) extends SendCommandGrpc.SendCommand with LogSupport {import gRPCServices._import PBConverter._var posConfig: com.typesafe.config.Config = _var exetimeout: Int = 5try {posConfig = ConfigFactory.load("pos.conf").getConfig("pos.server")exetimeout = posConfig.getInt("executimeout")}catch {case excp : Throwable =>log.warn(s"gRPCServices: ${excp.getMessage}")exetimeout = 5}override def singleResponse(request: PBPOSCommand): Future[PBPOSResponse] = {getPBResponse(writerRouter,request.toPOSComand, exetimeout)}override def multiResponse(request: PBPOSCommand, responseObserver: StreamObserver[PBPOSResponse]): Unit = ???
}object gRPCServices {import scala.concurrent.ExecutionContext.Implicits.globaldef getPBResponse(ref: ActorRef, cmd: POSMessage, executimeout: Int = 5): Future[PBPOSResponse] = {implicit val timeout = Timeout(executimeout second)val futRes: Future[POSResponse] = ask(ref, cmd).mapTo[POSResponse]futRes.map(_.toPBPOSResponse)}
}

现在需要把gRPCService与POS系统集成为一体，这样前端发来的PBCommand转换成Command后经POSAgent转发给集群分片writerRouter，writeRouter再发给writer去进行具体的操作处理，完后把POSResponse转换成PBPOSResponse通过service再返回前端：

  def getPBResponse(ref: ActorRef, cmd: POSMessage, executimeout: Int = 5): Future[PBPOSResponse] = {implicit val timeout = Timeout(executimeout second)val futRes: Future[POSResponse] = ask(ref, cmd).mapTo[POSResponse]futRes.map(_.toPBPOSResponse)}

可以看到上面使用了ask()模式来进行双向沟通。这个ref是一个中间信息交互actor （POSAgent）：

    var config = ConfigFactory.parseString("akka.remote.netty.tcp.port=\"" + port + "\"").withFallback(ConfigFactory.parseString("akka.remote.netty.tcp.hostname=\"" + host + "\"")).withFallback(ConfigFactory.parseString("cassandra-journal.contact-points=[\"" + host + "\"]")).withFallback(ConfigFactory.parseString("cassandra-snapshot-store.contact-points=[\"" + host + "\"]"))if (!seednodes.isEmpty)config = config.withFallback(ConfigFactory.parseString("akka.cluster.seed-nodes=[" + seednodes + "]"))//roles can be deployed on this nodeconfig = config.withFallback(ConfigFactory.parseString("akka.cluster.roles = [poswriter]")).withFallback(ConfigFactory.load())val posSystem = ActorSystem(systemName, config)posSystem.actorOf(ClusterMonitor.props, "cps-cluster-monitor")posSystem.actorOf(ActionReader.readerProps(showSteps),"reader")val readerRouter = posSystem.actorOf(ReaderRouter.props(showSteps),"reader-router")WriterShard.deployShard(posSystem)(ReaderInfo(readerRouter,writeOnly),showSteps)val posHandler = ClusterSharding(posSystem).shardRegion(WriterShard.shardName)val posref = posSystem.actorOf(WriterRouter.props(posHandler), "writer-router")val passer = posSystem.actorOf(POSAgent.props(posref), "pos-agent")val svc = SendCommandGrpc.bindService(new gRPCServices(passer), posSystem.dispatcher)runServer(svc)...package com.datatech.pos.cloudimport akka.actor._
import com.datatech.sdp
import sdp.logging._import Messages._
object POSAgent {def props(pos: ActorRef) = Props(new WriterRouter(pos))
}
class POSAgent(posHandler: ActorRef) extends Actor with LogSupport {var _sender: ActorRef = _override def receive: Receive = {case msg @ POSMessage(_,_) =>_sender = sender()posHandler ! msgcase resp: POSResponse  => _sender ! resp}
}...package com.datatech.pos.cloudimport akka.actor._
import com.datatech.sdp
import sdp.logging._import Messages._
object WriterRouter {def props(pos: ActorRef) = Props(new WriterRouter(pos))
}
class WriterRouter(posHandler: ActorRef) extends Actor with LogSupport {var _sender: ActorRef = _override def receive: Receive = {case msg @ POSMessage(_,_) =>_sender = sender()posHandler ! msgcase resp: POSResponse  => _sender ! resp
//      log.info(s"*********response from server: $resp *********")}
}

前端是gRPC的客户端。我们构建一个来测试后台控制逻辑：

package poc.clientimport scala.concurrent.Future
import com.datatech.pos.messages._
import com.datatech.pos.messages.SendCommandGrpc
import io.grpc.netty.{NegotiationType, NettyChannelBuilder}object POCClient {def main(args: Array[String]): Unit = {val channel = NettyChannelBuilder.forAddress("192.168.11.189",50051).negotiationType(NegotiationType.PLAINTEXT).build()/*//build connection channelval channel = io.grpc.ManagedChannelBuilder.forAddress("192.168.11.189",50051).usePlaintext(true).build()val pbCommand = PBPOSCommand(1022,"LogOn","888")//async callval asyncStub = SendCommandGrpc.blockingStub(channel)val futResponse: Future[PBPOSResponse] = asyncStub.singleResponse(pbCommand)import scala.concurrent.ExecutionContext.Implicits.globalfutResponse.foreach(result => println(result)) */val pbCommand = PBPOSCommand(1022,"LogOn","888")val syncStub1: SendCommandGrpc.SendCommandBlockingClient = SendCommandGrpc.blockingStub(channel)val response1: PBPOSResponse = syncStub1.singleResponse(pbCommand)println(s"${response1.msg}")val pbCommand2 = PBPOSCommand(1022,"LogOff","")//sync callval syncStub: SendCommandGrpc.SendCommandBlockingClient = SendCommandGrpc.blockingStub(channel)val response: PBPOSResponse = syncStub.singleResponse(pbCommand2)println(s"${response.msg}")scala.io.StdIn.readLine()channel.shutdown()}}

这里有几点必须注意：

1、protobuffer对象的强名称必须一致。在客户端用了同一个posmessages.proto定义文件：

syntax = "proto3";import "google/protobuf/wrappers.proto";
import "google/protobuf/any.proto";
import "scalapb/scalapb.proto";option (scalapb.options) = {// use a custom Scala package name// package_name: "io.ontherocks.introgrpc.demo"// don't append file name to packageflat_package: true// generate one Scala file for all messages (services still get their own file)single_file: true// add imports to generated file// useful when extending traits or using custom types// import: "io.ontherocks.hellogrpc.RockingMessage"// code to put at the top of generated file// works only with `single_file: true`//preamble: "sealed trait SomeSealedTrait"
};package com.datatech.pos.messages;message PBVchState {      //单据状态string opr  = 1;    //收款员int64  jseq = 2;    //begin journal sequence for read-side replayint32  num  = 3;    //当前单号int32  seq  = 4;    //当前序号bool   void = 5;    //取消模式bool   refd = 6;    //退款模式bool   susp = 7;    //挂单bool   canc = 8;    //废单bool   due  = 9;    //当前余额string su   = 10;   //主管编号string mbr  = 11;   //会员号int32  mode = 12;   //当前操作流程：0=logOff, 1=LogOn, 2=Payment
}message PBTxnItem {       //交易记录string txndate    = 1;   //交易日期string txntime    = 2;   //录入时间string opr        = 3;   //操作员int32  num        = 4;   //销售单号int32  seq        = 5;   //交易序号int32  txntype    = 6;   //交易类型int32  salestype  = 7;   //销售类型int32  qty        = 8;   //交易数量int32  price      = 9;   //单价（分）int32  amount     = 10;  //码洋（分）int32  disc       = 11;  //折扣率 (%)int32  dscamt     = 12;  //折扣额：负值  net实洋 = amount + dscamtstring member     = 13;  //会员卡号string code       = 14;  //编号（商品、卡号...）string acct       = 15;  //账号string dpt        = 16;  //部类
}message PBPOSResponse {int32  sts                  = 1;string msg                  = 2;PBVchState voucher          = 3;repeated PBTxnItem txnitems   = 4;}message PBPOSCommand {int64  shopid = 1;string commandname = 2;string delimitedparams = 3;
}service SendCommand {rpc SingleResponse(PBPOSCommand) returns (PBPOSResponse) {};rpc MultiResponse(PBPOSCommand) returns (stream PBPOSResponse) {};
}

注意package com.datatech.pos.messages, 这项在服务端和客户端必须一致。

2、另外就是客户端的channelbuilder：在scalaPB例子里使用的是ManagedChannelBuilder，这是一个实验阶段的东东：

    //build connection channelval channel = io.grpc.ManagedChannelBuilder.forAddress("132.232.229.60",50051).usePlaintext(true).build()

要用gRPC中正式的channelbuilder：

    val channel = NettyChannelBuilder.forAddress("192.168.11.189",50051).negotiationType(NegotiationType.PLAINTEXT).build()

上面这个NettyChannelBuilder的设置与那个io.grpc.ManagedChannelBuilder功能相等。但NettyChannelBuilder还具备更多的设置参数，如ssl/tls设置。

3、还有：因为客户端是按照顺序来发送操作指令的，每发一个指令，等待返回结果后才能再发下一个指令。所以必须使用同步客户端调用函数blockingStub。

下面是本次示范的一些配置文档：

project/plugins.sbt

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.9")
addSbtPlugin("net.virtual-void" % "sbt-dependency-graph" % "0.9.2")
addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.3.15")
addSbtPlugin("com.thesamet" % "sbt-protoc" % "0.99.21")
libraryDependencies += "com.thesamet.scalapb" %% "compilerplugin" % "0.9.0-M6"

build.sbt

name := "pos-on-cloud"version := "0.1"scalaVersion := "2.12.8"scalacOptions += "-Ypartial-unification"val akkaVersion = "2.5.23"libraryDependencies := Seq("com.typesafe.akka" %% "akka-cluster-metrics" % akkaVersion,"com.typesafe.akka" %% "akka-cluster-sharding" % akkaVersion,"com.typesafe.akka" %% "akka-persistence" % akkaVersion,"com.lightbend.akka" %% "akka-stream-alpakka-cassandra" % "1.0.1","org.mongodb.scala" %% "mongo-scala-driver" % "2.6.0","com.lightbend.akka" %% "akka-stream-alpakka-mongodb" % "1.0.1","com.typesafe.akka" %% "akka-persistence-query" % akkaVersion,"com.typesafe.akka" %% "akka-persistence-cassandra" % "0.97","com.datastax.cassandra" % "cassandra-driver-core" % "3.6.0","com.datastax.cassandra" % "cassandra-driver-extras" % "3.6.0","ch.qos.logback"  %  "logback-classic"   % "1.2.3","io.monix" %% "monix" % "3.0.0-RC2","org.typelevel" %% "cats-core" % "2.0.0-M1","io.grpc" % "grpc-netty" % scalapb.compiler.Version.grpcJavaVersion,"com.thesamet.scalapb" %% "scalapb-runtime" % scalapb.compiler.Version.scalapbVersion % "protobuf","com.thesamet.scalapb" %% "scalapb-runtime-grpc" % scalapb.compiler.Version.scalapbVersion)PB.targets in Compile := Seq(scalapb.gen() -> (sourceManaged in Compile).value
)enablePlugins(JavaAppPackaging)

resouces/application.conf

akka.actor.warn-about-java-serializer-usage = off
akka.log-dead-letters-during-shutdown = off
akka.log-dead-letters = off
akka.remote.use-passive-connections=offakka {loglevel = INFOactor {provider = "cluster"}remote {log-remote-lifecycle-events = onnetty.tcp {hostname = "127.0.0.1"# port set to 0 for netty to randomly choose fromport = 0}}cluster {seed-nodes = ["akka.tcp://cloud-pos-server@172.27.0.8:2551","akka.tcp://cloud-pos-server@172.27.0.7:2551"]log-info = offsharding {role = "poswriter"passivate-idle-entity-after = 30 m}}persistence {journal.plugin = "cassandra-journal"snapshot-store.plugin = "cassandra-snapshot-store"}}cassandra-journal {contact-points = ["172.27.0.8","172.27.0.7","172.27.0.15"]
}cassandra-snapshot-store {contact-points = ["172.27.0.8","172.27.0.7","172.27.0.15"]
}# Enable metrics extension in akka-cluster-metrics.
akka.extensions=["akka.cluster.metrics.ClusterMetricsExtension"]akka.actor.deployment {/reader-router/readerRouter = {# Router type provided by metrics extension.router = cluster-metrics-adaptive-group# Router parameter specific for metrics extension.# metrics-selector = heap# metrics-selector = load# metrics-selector = cpumetrics-selector = mix#routees.paths = ["/user/reader"]cluster {max-nr-of-instances-per-node = 10max-total-nr-of-instances = 1000enabled = on#set to on when there is a instance of routee created#on the same node as the router#very important to set this off, could cause missing msg in local clusterallow-local-routees = on}}
}dbwork-dispatcher {# Dispatcher is the name of the event-based dispatchertype = Dispatcher# What kind of ExecutionService to useexecutor = "fork-join-executor"# Configuration for the fork join poolfork-join-executor {# Min number of threads to cap factor-based parallelism number toparallelism-min = 2# Parallelism (threads) ... ceil(available processors * factor)parallelism-factor = 2.0# Max number of threads to cap factor-based parallelism number toparallelism-max = 10}# Throughput defines the maximum number of messages to be# processed per actor before the thread jumps to the next actor.# Set to 1 for as fair as possible.throughput = 100
}

resources/logback.xml

<?xml version="1.0" encoding="UTF-8"?>
<configuration><appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"><encoder><Pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n</Pattern></encoder></appender><logger name="com.datatech" level="info"additivity="false"><appender-ref ref="STDOUT" /></logger><logger name="com.datatech.sdp" level="info"additivity="false"><appender-ref ref="STDOUT" /></logger><root level="warn"><appender-ref ref="STDOUT" /></root>
</configuration>

resources/pos.conf

pos {server {debug = falsecqlport = 9042readinterval = 1000executimeout = 5}
}

Akka-CQRS（9）- gRPC，实现前端设备与平台系统的高效集成相关推荐

化工园区应急预警联动合成作战平台系统解决方案
方案概述化工园区应急预警联动合成作战平台系统是基于云计算.物联网技术.移动互联网技术.GPS服务为一体的专门适用于在化工园区人员紧急求救的视频联网报警联动和指挥调度平台. 此平台集视频报警.语音报警 ...
MindSpore平台系统类
MindSpore平台系统类 Q:MindSpore只能在华为自己的NPU上跑么? A: MindSpore同时支持华为自己的Ascend NPU.GPU与CPU,是支持异构算力的. Q:MindSp ...
一个平台系统架构师的能力模型是啥
之前的文章分享过我自己琢磨的一个技术高P的能力模型(参见:高P的能力模型),其中的一个方向是cover端到端解决的能力.那怎么叫端到端解决方案的能力呢? 我自己想了下,一种切入端到端解决方案能力的方式 ...
XXX管理平台系统——项目教训
XXX管理平台系统项目教训前言闲来无事聊一下自己的教训吧,经验也是在教训中不断成长的. 技术方面之前对硬件和网络缺乏基本的选型概念,以及对整个系统的整体和技术方案把握有所欠缺,导致整个系统架构 ...
XXX管理平台系统——会议管理
XXX管理平台系统会议管理关于项目中的会议管理,其实并无定数可言,与个人的爱好.管理风格.沟通能力.团队组织.项目规模.甚至公司的IT管理风格息息相关. 在H公司工作的时候,我的departmen ...
深入浅出 - Android系统移植与平台开发（十）- Android编译系统与定制Android平台系统（瘋耔修改篇二）...
第四章.Android编译系统与定制Android平台系统 4.1Android编译系统 Android的源码由几十万个文件构成,这些文件之间有的相互依赖,有的又相互独立,它们按功能或类型又被放到不同 ...
ZLMS教学管理平台系统V1.2.0最新版本发布，支持纯Web视频直播点播，还带运营在线支付功能！完全免费提供！...
ZLMS教学管理平台系统V1.2.0最新版本发布,支持纯Web视频直播,点播!还带在线支付功能! ZLMS 开发团队在综合参考了各方面的合理建议之后,经过两个多月的紧张开发及测试,终于发布V1.2.0 ...
XXX管理平台系统——概要
XXX管理平台系统概要项目背景某公司为了解决各部门信息孤岛效应,为了向客户提供具有公司品牌的.内容全面.高质量.个性化.统一的优质信息服务,树立公司形象.提高客户忠诚度,同时打造一个优质的客户品 ...
XXX管理平台系统——架构
XXX管理平台系统架构前言系统架构是项目中技术实现的最重要的环节.系统架构的良好与否关系到系统的性能指标.安全指标.稳定性指标.可扩展性.业务实现等等. 系统架构涉及到系统硬件的选型.网络拓扑. ...
大型动态应用系统平台系统架构？这些大家并不陌生
大型动态应用系统平台系统架构[多图] 动态应用,是相对于网站静态内容而言,是指以c/c++.php.Java.perl..net等服务器端语言开发的网络应用软件,比如论坛.网络相册.交友.BLOG等常 ...

Akka-CQRS（9）- gRPC，实现前端设备与平台系统的高效集成

Akka-CQRS（9）- gRPC，实现前端设备与平台系统的高效集成相关推荐

最新文章

热门文章