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。

相关文章

BUUCTF靶场[web][极客大挑战 2019]Http、[HCTF 2018]admin

BUUCTF靶场[web][极客大挑战 2019]Http、[HCTF 2018]admin

目录   [web][极客大挑战 2019]Http 考点：Referer协议、UA协议、X-Forwarded-For协议 [web][HCTF 2018]admin 考点：弱密码字典爆破 四种方法：   [web][极客大挑战 2019]Http 考点：Referer协议、UA协议、X-Forwarded-For协议 访问环境 老规矩，我们先查看源代码
阅读更多...
【Linux】应用层http协议

【Linux】应用层http协议

一、HTTP协议 1.1 简要介绍一下HTTP        我们在网络的应用层中可以自己定义协议，但是，已经有大佬定义了一些现成的，非常好用的应用层协议，供我们直接使用，HTTP（超文本传输协议）就是其中之一。        在互联网世界中，HTTP（超文本传输协议）是一个至关重要的协议，他定义了客户端（如浏览器）与服务器之间如何进行通信，以交换或者传输超文本（比如HTML文档）。
阅读更多...
如何确定 Go 语言中 HTTP 连接池的最佳参数？

如何确定 Go 语言中 HTTP 连接池的最佳参数？

确定 Go 语言中 HTTP 连接池的最佳参数可以通过以下几种方式： 一、分析应用场景和需求 并发请求量： 确定应用程序在特定时间段内可能同时发起的 HTTP 请求数量。如果并发请求量很高，需要设置较大的连接池参数以满足需求。例如，对于一个高并发的 Web 服务，可能同时有数百个请求在处理，此时需要较大的连接池大小。可以通过压力测试工具模拟高并发场景，观察系统在不同并发请求下的性能表现，从而
阅读更多...
Anaconda 中遇到CondaHTTPError: HTTP 404 NOT FOUND for url的问题及解决办法

Anaconda 中遇到CondaHTTPError: HTTP 404 NOT FOUND for url的问题及解决办法

最近在跑一个开源项目遇到了以下问题，查了很多资料都大（抄）同（来）小（抄）异（去）的，解决不了根本问题，费了很大的劲终于得以解决，记录如下： 1、问题及过程： (myenv) D:\Workspace\python\XXXXX>conda install python=3.6.13 Solving environment: done.....Proceed ([y]/n)? yDownloa
阅读更多...
构建高性能WEB之HTTP首部优化

构建高性能WEB之HTTP首部优化

0x00 前言 在讨论浏览器优化之前，首先我们先分析下从客户端发起一个HTTP请求到用户接收到响应之间，都发生了什么？知己知彼，才能百战不殆。这也是作为一个WEB开发者，为什么一定要深入学习TCP/IP等网络知识。 0x01 到底发生什么了？ 当用户发起一个HTTP请求时，首先客户端将与服务端之间建立TCP连接，成功建立连接后，服务端将对请求进行处理，并对客户端做出响应，响应内容一般包括响应
阅读更多...
Golang支持平滑升级的HTTP服务

Golang支持平滑升级的HTTP服务

前段时间用Golang在做一个HTTP的接口，因编译型语言的特性，修改了代码需要重新编译可执行文件，关闭正在运行的老程序，并启动新程序。对于访问量较大的面向用户的产品，关闭、重启的过程中势必会出现无法访问的情况，从而影响用户体验。 使用Golang的系统包开发HTTP服务，是无法支持平滑升级（优雅重启）的，本文将探讨如何解决该问题。 一、平滑升级（优雅重启）的一般思路 一般情况下，要实现平滑
阅读更多...
Java http请求示例

Java http请求示例

使用HttpURLConnection public static String httpGet(String host) {HttpURLConnection connection = null;try {URL url = new URL(host);connection = (HttpURLConnection) url.openConnection();connection.setReq
阅读更多...
3.比 HTTP 更安全的 HTTPS（工作原理理解、非对称加密理解、证书理解）

3.比 HTTP 更安全的 HTTPS（工作原理理解、非对称加密理解、证书理解）

所谓的协议 协议只是一种规则，你不按规则来就无法和目标方进行你的工作 协议说白了只是人定的规则，任何人都可以定协议 我们不需要太了解细节，这些制定和完善协议的人去做的，我们只需要知道协议的一个大概 HTTPS 协议 1、概述 HTTPS（Hypertext Transfer Protocol Secure）是一种安全的超文本传输协议，主要用于在客户端和服务器之间安全地传输数据
阅读更多...
IEEE会议投稿资料汇总http://cadcg2015.nwpu.edu.cn/index.htm

IEEE会议投稿资料汇总http://cadcg2015.nwpu.edu.cn/index.htm

最近投了篇IEEE的顶级会议文章，一下是比较有用的一些资料，以供参考。 1.会议主页：http://cadcg2015.nwpu.edu.cn/index.htm     （The 14th International Conference on Computer-Aided Design and Computer Graphics (CAD/Graphics 2015)） 2.I
阅读更多...
Thor与Http catcher介绍

Thor与Http catcher介绍

IOS破解VIP神器Thor和Http catcher，大家都听说过吗？ 简单聊聊两款软件的区别和使用体验： Thor： （1）UI界面很直观，清晰，可以直接选择过滤器 （2）创建过滤器对于新手来说个人感觉比Http catcher难上手一点 （3）无法同时选择多个过滤器 Http catcher： （1）第一次使用找不到过滤器（重写）在哪里 （2）对于新手，只要会一点正则表达式基
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2