akka-grpc - 基于akka-http和akka-streams的scala gRPC开发工具

2024-04-09 04:32
文章标签 http scala grpc 开发工具 akka streams

本文主要是介绍akka-grpc - 基于akka-http和akka-streams的scala gRPC开发工具，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

 关于grpc，在前面的scalaPB讨论里已经做了详细的介绍：google gRPC是一种全新的RPC框架，在开源前一直是google内部使用的集成工具。gRPC支持通过http/2实现protobuf格式数据交换。protobuf即protocol buffer，是google发明的一套全新的序列化传输协议serialization-protocol，是二进制编码binary-encoded的，相对java-object，XML，Json等在空间上占有优势，所以数据传输效率更高。由于gRPC支持http/2协议，可以实现双向通讯duplex-communication，解决了独立request/response交互模式在软件编程中的诸多局限。这是在系统集成编程方面相对akka-http占优的一个亮点。protobuf格式数据可以很方便的转换成 json格式数据，支持对外部系统的的开放协议数据交换。这也是一些人决定选择gRPC作为大型系统微服务集成开发工具的主要原因。更重要的是：用protobuf和gRPC进行client/server交互不涉及任何http对象包括httprequest,httpresponse，很容易上手使用，而且又有在google等大公司内部的成功使用经验，用起来会更加放心。

虽然gRPC基于http/2协议在网络通讯效率和模式上有了很大的提升，但近段时间对gRPC使用的调研主要还是冲着protobuf来的。在http/1应用中对二进制文件的传输交换有诸多限制和不便，特别是效率方面的问题。在protobuf这种序列化模式中对任何类型的数据格式都一视同仁，可以很方便的实现图片等文件的上传下载。另一个原因是：http/2并不是一种普及的协议，并不适合作为一个开放数据平台的连接协议。再就是：虽然gRPC是基于http协议上的，但对于HttpRequest的调用却非常不便，需要通过interceptor来实现，不但麻烦而且有门槛。

实际上，在使用scalaPB的过程中一直在关注akka-grpc的发展，直到v1.01发布。这是一个正式版本，相信不会在模式、风格、语法上再有大的改变，应该值得试着使用了。

对akka-http用户来说，akka-grpc具有很大吸引（相对其它gRPC开放工具），因为它是基于akka-http的，看看下面grpc服务端的接口：

    // Bind service handler servers to localhost:8080/8081val binding = Http().bindAndHandleAsync(serviceHandlers,interface = host,port = port,connectionContext = HttpConnectionContext())// report successful bindingbinding.foreach { binding => println(s"gRPC server bound to: ${binding.localAddress}") }binding

上面这段代码不就是akka-http里的吗。那么可以想象得到如果需要支持http+rpc混合模式的应用，akka-grpc将会发挥很大作用，这也是akka-http下一步的发展趋势。

至于akka-grpc基于akka-streams的特性，我并没有感到太大的兴趣。如上所述，我们的目标是实现一种开放数据平台的终端接入接口。akka-streams恰恰把总体方案限制在了内部系统集成模式，因为服务端客户端就是akka-streams的两端，是内部系统集成的场景。也许，在开发一套内部IT系统的过程中akka-grpc可以很趁手。

与scalaPB一样，akka-grpc也是通过编译IDL（.proto）文件用相应的插件（plugin）产生相关的scala类和服务函数代码。实际上akka-grpc产生代码的plugin还是采用scalaPB的插件，这个过程已经在scalaPB系列博客里详细介绍过了。

gRPC支持下面四种交互协议：

1、Unary：独立的一对client-request/server-response，是我们常用的http交互模式

2、Server-Streaming：client发出一个request后从server端接收一串多个response

3、Client-Streaming：client向server发送一串多个request后从server接收一个response

4、Bidirectional-Streaming：还是由client首先发送request启动连接，然后在这个连接上client/server双方可以不断交互信息。

在akka-grpc的官网上有很好的示范例子。我在例子的基础上增加了身份验证使用的示范。数据类型和服务函数用IDL定义的.proto文件内容如下：

syntax = "proto3";//#options
option java_multiple_files = true;
//option java_package = "learn.akka.grpc";
//option java_outer_classname = "GreeterProto";package learn.akka.grpc;
//#options//#services
// The greeting service definition.
service GreeterService {//// Sends a greeting //*****///      HELLO       //*****/rpc SayHello (HelloRequest) returns (HelloReply) {}// Comment spanning// on several linesrpc ItKeepsTalking (stream HelloRequest) returns (HelloReply) {}/** C style comments*/rpc ItKeepsReplying (HelloRequest) returns (stream HelloReply) {}/* C style comments* on several lines* with non-empty heading/trailing line */rpc StreamHellos (stream HelloRequest) returns (stream HelloReply) {}
}
service AuthService {rpc GetAuthToken(UCredential) returns (JWToken) {}
}
//#services//#messages
// The request message containing the user's name.
message HelloRequest {string name = 1;
}// The response message containing the greetings
message HelloReply {string message = 1;
}message UCredential {string userid = 1;string password = 2;
}message JWToken {string jwt = 1;
}
//#messages

注意：文件里增加了AuthService服务，函数和类型分别为：GetAuthToken, UCredential, JWToken

身份验证需要通过HttpHeader实现，所以产生的函数样板里必须提供使用HttpHeader的参数。这个可以通过在built.sbt里增加akkaGrpcCodeGeneratorSettings += "server_power_apis"就能做到。下面是在产生的sayHello函数款式基础上的实现示范代码：

  override def sayHello(in: HelloRequest,header: Metadata): Future[HelloReply] = {if (isAuthenticated(header)) {println(s"sayHello to ${in.name}")FastFuture.successful(HelloReply(s"Hello, ${in.name}"))} else {println(s"${in.name} Unauthorized!")FastFuture.successful(HelloReply(s"Unauthorized! ${in.name}"))}}

header是客户端提交的HttpRequest，如下：

    //#with-metadatadef singleRequestReply(jwt: String): Unit = {sys.log.info("Performing request")val reply = client.sayHello().addHeader("Bearer", jwt).invoke(HelloRequest("Alice"))println(s"got single reply: ${Await.result(reply, 5.seconds).message}")}

客户端向服务端提供了JWT header 作为身份凭证。

上面提到，虽然http/2推出已经不短时间了，但尚未得到普及性的认可。即使是低点版本的websocket，也只能在一小撮专业的应用中得到使用。所以，akka-grpc并没有提供对OAuth2规范身份验证的支持。在这个例子里我们就只能进行基本的身份证明（如店号、机器号等），但身份验证过程的安全性就不做任何加密操作了。首先，在IDL文件里增加对AuthService的描述，如下：

service AuthService {rpc GetAuthToken(UCredential) returns (JWToken) {}
}message UCredential {string userid = 1;string password = 2;
}message JWToken {string jwt = 1;
}

现在需要在系统里提供两套服务：authService和greeterService, 如下：

class gRPCServer(host: String, port: Int) {def run(system: ActorSystem[_]): Future[Http.ServerBinding] = {implicit val classic = system.toClassicimplicit val ec: ExecutionContext = system.executionContextval greeterService: PartialFunction[HttpRequest, Future[HttpResponse]] =learn.akka.grpc.GreeterServicePowerApiHandler.partial(new GreeterServices(system))val authService: PartialFunction[HttpRequest, Future[HttpResponse]] =learn.akka.grpc.AuthServicePowerApiHandler.partial(new AuthServices())val serviceHandlers: HttpRequest => Future[HttpResponse] =ServiceHandler.concatOrNotFound(greeterService, authService)// Bind service handler servers to localhost:8080/8081val binding = Http().bindAndHandleAsync(serviceHandlers,interface = host,port = port,connectionContext = HttpConnectionContext())// report successful bindingbinding.foreach { binding => println(s"gRPC server bound to: ${binding.localAddress}") }binding//#server}
}

获取身份凭证jwt示范如下：

客户端：val authenticator = AuthServiceClient(clientSettings)val futJwt = authenticator.getAuthToken(UCredential("9013","123456"))val jwt = Await.result(futJwt,5.seconds).jwtprintln(s"got jwt: ${jwt}")服务端：class AuthServices() extends AuthServicePowerApi {//#service-request-replyoverride def getAuthToken(in: UCredential, header: Metadata): Future[JWToken] =FastFuture.successful(JWToken("jwtstring"))
}

服务端的身份验证示范：

  override def sayHello(in: HelloRequest,header: Metadata): Future[HelloReply] = {if (isAuthenticated(header)) {println(s"sayHello to ${in.name}")FastFuture.successful(HelloReply(s"Hello, ${in.name}"))} else {println(s"${in.name} Unauthorized!")FastFuture.successful(HelloReply(s"Unauthorized! ${in.name}"))}}private def isAuthenticated(metadata: Metadata): Boolean = {metadata.getText("bearer") match {case Some(t) => t == "jwtstring"case _ => false}}

好了，下面是本次示范的源代码：

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.lightbend.akka.grpc" % "sbt-akka-grpc" % "1.0.1")
addSbtPlugin("com.lightbend.sbt" % "sbt-javaagent" % "0.1.5")

build.sbt

val AkkaVersion = "2.6.8"
val AkkaPersistenceCassandraVersion = "1.0.1"
val AkkaHttpVersion = "10.1.12"
val elastic4sVersion = "7.6.0"
lazy val akkaGrpcVersion = "1.0.1"
enablePlugins(AkkaGrpcPlugin)
lazy val `learn-akka-grpc` = project.in(file(".")).settings(organization := "com.datatech",version := "1.0",scalaVersion := "2.13.2",scalacOptions in Compile ++= Seq("-deprecation", "-feature", "-unchecked", "-Xlog-reflective-calls", "-Xlint"),javacOptions in Compile ++= Seq("-Xlint:unchecked", "-Xlint:deprecation"),libraryDependencies ++= Seq("com.typesafe.akka" %% "akka-actor-typed" % AkkaVersion,"com.typesafe.akka" %% "akka-stream" % AkkaVersion,"com.typesafe.akka" %% "akka-discovery" % AkkaVersion,"com.typesafe.akka" %% "akka-cluster-sharding-typed" % AkkaVersion,"com.typesafe.akka" %% "akka-persistence-typed" % AkkaVersion,"com.typesafe.akka" %% "akka-persistence-query" % AkkaVersion,"com.typesafe.akka" %% "akka-serialization-jackson" % AkkaVersion,"com.typesafe.akka" %% "akka-persistence-cassandra" % AkkaPersistenceCassandraVersion,"com.typesafe.akka" %% "akka-http" % AkkaHttpVersion,"com.typesafe.akka" %% "akka-http-spray-json" % AkkaHttpVersion,"com.typesafe.akka" %% "akka-slf4j" % AkkaVersion,"com.datastax.cassandra" % "cassandra-driver-core" % "3.6.0","com.datastax.cassandra" % "cassandra-driver-extras" % "3.6.0","com.sksamuel.elastic4s" %% "elastic4s-client-esjava" % elastic4sVersion,"ch.qos.logback" % "logback-classic" % "1.2.3","com.pauldijou" %% "jwt-core" % "3.0.1","org.json4s" %% "json4s-native" % "3.7.0-M4"))enablePlugins(JavaAppPackaging)
akkaGrpcCodeGeneratorSettings += "server_power_apis"

gRPCServer.scala

package learn.akka.grpc.demo
import akka.actor.typed._
import akka.actor.typed.scaladsl._
import akka.actor.typed.scaladsl.adapter._
import akka.http.scaladsl._
import com.typesafe.config.ConfigFactory
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import scala.concurrent._
import akka.grpc.scaladsl.ServiceHandler
//#import//#server
object gRPCServer {def main(args: Array[String]): Unit = {val hostPat = "(.*):(.*)".rval (host, port) = args(0) match {case hostPat(h, p) => (h, p)case _ => ("localhost", "8080")}// important to enable HTTP/2 in ActorSystem's configval conf = ConfigFactory.parseString("akka.http.server.preview.enable-http2 = on").withFallback(ConfigFactory.defaultApplication())val system = ActorSystem[Nothing](Behaviors.empty, "grpc-server", conf)new gRPCServer(host,port.toInt).run(system)}
}class gRPCServer(host: String, port: Int) {def run(system: ActorSystem[_]): Future[Http.ServerBinding] = {implicit val classic = system.toClassicimplicit val ec: ExecutionContext = system.executionContextval greeterService: PartialFunction[HttpRequest, Future[HttpResponse]] =learn.akka.grpc.GreeterServicePowerApiHandler.partial(new GreeterServices(system))val authService: PartialFunction[HttpRequest, Future[HttpResponse]] =learn.akka.grpc.AuthServicePowerApiHandler.partial(new AuthServices())val serviceHandlers: HttpRequest => Future[HttpResponse] =ServiceHandler.concatOrNotFound(greeterService, authService)// Bind service handler servers to localhost:8080/8081val binding = Http().bindAndHandleAsync(serviceHandlers,interface = host,port = port,connectionContext = HttpConnectionContext())// report successful bindingbinding.foreach { binding => println(s"gRPC server bound to: ${binding.localAddress}") }binding//#server}
}

GreeterServices.scala

package learn.akka.grpc.demo
import akka.NotUsed
import akka.actor.typed.ActorSystem
import akka.http.scaladsl.util.FastFuture
import akka.grpc.scaladsl._
import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.Source
import learn.akka.grpc._import scala.concurrent.{ExecutionContext, Future}//#import//#service-request-reply
//#service-stream
class GreeterServices(system: ActorSystem[_]) extends GreeterServicePowerApi {private implicit val sys: ActorSystem[_] = systemprivate implicit val ec: ExecutionContext = system.executionContextoverride def sayHello(in: HelloRequest,header: Metadata): Future[HelloReply] = {if (isAuthenticated(header)) {println(s"sayHello to ${in.name}")FastFuture.successful(HelloReply(s"Hello, ${in.name}"))} else {println(s"${in.name} Unauthorized!")FastFuture.successful(HelloReply(s"Unauthorized! ${in.name}"))}}override def itKeepsTalking(in: Source[HelloRequest, NotUsed],header: Metadata): Future[HelloReply] = {if (isAuthenticated(header)) {println(s"sayHello to in stream...")in.runWith(Sink.seq).map(elements => HelloReply(s"Hello, ${elements.map(_.name).mkString(", ")}"))} else {println(s"Unauthorized!")in.runWith(Sink.seq).map(elements => HelloReply(s"Unauthorized, ${elements.map(_.name).mkString(", ")}"))}}override def itKeepsReplying(in: HelloRequest,header: Metadata): Source[HelloReply, NotUsed] = {if (isAuthenticated(header)) {println(s"sayHello to ${in.name} with stream of chars...")Source(s"Hello, ${in.name}".toList).map(character => HelloReply(character.toString))} else {println(s"${in.name} Unauthorized!")Source(s"Unauthorized, ${in.name}".toList).map(character => HelloReply(character.toString))}}override def streamHellos(in: Source[HelloRequest, NotUsed],header: Metadata): Source[HelloReply, NotUsed] = {if (isAuthenticated(header)) {println(s"sayHello to stream...")in.map(request => HelloReply(s"Hello, ${request.name}"))} else {println(s"Unauthorized!")in.map(request => HelloReply(s"Unauthorized, ${request.name}"))}}private def isAuthenticated(metadata: Metadata): Boolean = {metadata.getText("bearer") match {case Some(t) => t == "jwtstring"case _ => false}}
}
//#service-stream
//#service-request-reply

AuthServices.scala

package learn.akka.grpc.demo
import akka.http.scaladsl.util.FastFuture
import akka.grpc.scaladsl._
import learn.akka.grpc._import scala.concurrent.Futureclass AuthServices() extends AuthServicePowerApi {//#service-request-replyoverride def getAuthToken(in: UCredential, header: Metadata): Future[JWToken] =FastFuture.successful(JWToken("jwtstring"))
}

AuthServiceClient.scala

package learn.akka.grpc.demo
import scala.concurrent.Await
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.Tryimport akka.Done
import akka.NotUsed
import akka.actor.ActorSystem
import akka.grpc.GrpcClientSettings
import akka.stream.scaladsl.Sourceimport learn.akka.grpc._object AuthGreeterClient {def main(args: Array[String]): Unit = {implicit val sys = ActorSystem("grpc-client")implicit val ec = sys.dispatcherval clientSettings = GrpcClientSettings.fromConfig(GreeterService.name)val client = GreeterServiceClient(clientSettings)val authenticator = AuthServiceClient(clientSettings)val futJwt = authenticator.getAuthToken(UCredential("9013","123456"))val jwt = Await.result(futJwt,5.seconds).jwtprintln(s"got jwt: ${jwt}")singleRequestReply(jwt)streamingRequest(jwt)streamingReply(jwt)streamingRequestReply(jwt)sys.scheduler.scheduleAtFixedRate(1.second, 1.second)(new Runnable {override def run(): Unit = Try(singleRequestReply(jwt))})//#with-metadatadef singleRequestReply(jwt: String): Unit = {sys.log.info("Performing request")val reply = client.sayHello().addHeader("Bearer", jwt).invoke(HelloRequest("Alice"))println(s"got single reply: ${Await.result(reply, 5.seconds).message}")}//#with-metadatadef streamingRequest(jwt: String): Unit = {val requests = List("Alice", "Bob", "Peter").map(HelloRequest(_))val reply = client.itKeepsTalking().addHeader("Bearer", jwt).invoke(Source(requests))println(s"got single reply for streaming requests: ${Await.result(reply, 5.seconds).message}")}def streamingReply(jwt: String): Unit = {val responseStream = client.itKeepsReplying().addHeader("Bearer", jwt).invoke(HelloRequest("Alice"))val done: Future[Done] =responseStream.runForeach(reply => println(s"got streaming reply: ${reply.message}"))Await.ready(done, 1.minute)}def streamingRequestReply(jwt: String): Unit = {val requestStream: Source[HelloRequest, NotUsed] =Source.tick(100.millis, 1.second, "tick").zipWithIndex.map { case (_, i) => i }.map(i => HelloRequest(s"Alice-$i")).take(10).mapMaterializedValue(_ => NotUsed)val responseStream: Source[HelloReply, NotUsed] =client.streamHellos().addHeader("Bearer", jwt).invoke(requestStream)val done: Future[Done] =responseStream.runForeach(reply => println(s"got streaming reply: ${reply.message}"))Await.ready(done, 1.minute)}}
}

 

这篇关于akka-grpc - 基于akka-http和akka-streams的scala gRPC开发工具的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

微服务中RPC的强类型检查与HTTP的弱类型对比

微服务中RPC的强类型检查与HTTP的弱类型对比

在微服务架构中，服务间的通信是一个至关重要的环节。其中，远程过程调用（RPC）和HTTP是两种最常见的通信方式。虽然它们都能实现服务间的数据交换，但在类型检查方面，RPC的强类型检查和HTTP的弱类型之间有着显著的差异。本文将深入探讨这两种通信方式在类型检查方面的优缺点，以及它们对微服务架构的影响。 一、RPC的强类型检查 RPC的强类型检查是其核心优势之一。在RPC通信中，客户端和服务端都使
阅读更多...
el-upload 上传图片及回显照片和预览图片,文件流和http线上链接格式操作

el-upload 上传图片及回显照片和预览图片,文件流和http线上链接格式操作

<div v-for="(info, index) in zsjzqwhxqList.helicopterTourInfoList" :key="info.id" >编辑上传图片// oss返回线上地址http链接格式：<el-form-itemlabel="巡视结果照片":label-width="formLabelWidth"><el-upload:action="'http:
阅读更多...
HTTP状态码中301与302的区别

HTTP状态码中301与302的区别

一．官方说法  301，302 都是HTTP状态的编码，都代表着某个URL发生了转移，不同之处在于：  301 redirect: 301 代表永久性转移(Permanently Moved)。  302 redirect: 302 代表暂时性转移(Temporarily Moved )。  这是很官方的说法，那么它们的区别到底是什么呢？  1.1、什么是301转向?什么是301重定向?
阅读更多...
web前端不可错过的开发工具–Adobe Brackets（开源、简洁强大的HTML、CSS和JavaScript集成开发环境）

web前端不可错过的开发工具–Adobe Brackets（开源、简洁强大的HTML、CSS和JavaScript集成开发环境）

Adobe Brackets是一个开源的基于HTML/CSS/JavaScript开发，运行在native shell上的集成开发环境。该项目由Adobe创建和维护，根据MIT许可证发布。提供Windows和OS X平台支持。 Brackets的特点是简约、快捷，没有很多的视图或者面板，它的核心目标是减少在开发过程中那些效率低下的重复性工作，例如浏览器刷新，修改元素的样式，搜索功能等等。
阅读更多...
使用 axios 进行 HTTP 请求

使用 axios 进行 HTTP 请求

使用 axios 进行 HTTP 请求 文章目录 使用 axios 进行 HTTP 请求1、介绍2、安装和引入3、axios 基本使用4、axios 发送 GET 请求5、axios 发送 POST 请求6、高级使用7、总结 1、介绍 什么是 axios axios 是一个基于 promise 的 HTTP 库，可以用于浏览器和 Node.js 中发送 HTT
阅读更多...
HTTP基本概念介绍

HTTP基本概念介绍

HTTP概述 HTTP : 超文本传输协议，HTTP是浏览器端Web通信的基础。 一， 两种架构 B/S架构：Browser/Server，浏览器/服务器架构。 B:  浏览器，比如Firefox 、Google 、Internet; S:  服务器，Apache，nginx； C/S架构：Client/Server，客户端/服务器架构。 B/S架构相对于C/S架构，客户机上无需安装任何软件
阅读更多...
网络编程篇:HTTP协议

网络编程篇:HTTP协议

一.预备知识 在客户端访问服务端时，要用ip+port，但是在日常用户访问服务端的时候，并不会直接使用ip，而是使用域名，比如：百度(www.baidu,com)。                                                                                    在浏览器处理时，会将域名解析为对应的ip+port。
阅读更多...
FFplay源码分析-streams_open

FFplay源码分析-streams_open

《FFmpeg原理》的社群来了，想加入社群的朋友请购买 VIP 版，VIP 版有更高级的内容与答疑服务。 本系列 以 ffmpeg4.2 源码为准，下载地址：链接：百度网盘 提取码：g3k8 FFplay 源码分析系列以一条简单的命令开始，ffplay -i a.mp4。a.mp4下载链接：百度网盘，提取码：nl0s 。 如下图所示，本文主要讲解 streams_open() 函数的
阅读更多...
基于http协议的接口测试(2)

基于http协议的接口测试(2)

一）.接口约定： 1、接口均采用http协议，Get方法。 2、城市公交请求参数为app_key、city、start_addr、end_addr，app_key为需要你申请密钥key http://openapi.aibang.com/bus/transfer?app_key=????&city=????&start_addr=????&end_addr=???? 如果不清楚请求
阅读更多...
http:与https:到底有哪些区别？

http:与https:到底有哪些区别？

http和https使用的是完全不同的连接方式,用的端口也不一样,前者是80,后者是443。http的连接很简单,是无状态的,... HTTPS协议是由SSL+HTTP协议构建的可进行加密传输、身份认证的网络协议要比http协议安全
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2