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/8081

    val binding = Http().bindAndHandleAsync(

      serviceHandlers,

      interface = host,

      port = port,

      connectionContext = HttpConnectionContext())

    // report successful binding

    binding.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 lines

  rpc 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 = ;

}

// The response message containing the greetings

message HelloReply {

  string message = ;

}

message UCredential {

  string userid = ;

  string password = ;

}

message JWToken {

  string jwt = ;

}

//#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-metadata

    def 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 = ;

  string password = ;

}

message JWToken {

  string jwt = ;

}

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

class gRPCServer(host: String, port: Int) {

  def run(system: ActorSystem[_]): Future[Http.ServerBinding] = {

    implicit val classic = system.toClassic

    implicit val ec: ExecutionContext = system.executionContext

    val 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/8081

    val binding = Http().bindAndHandleAsync(

      serviceHandlers,

      interface = host,

      port = port,

      connectionContext = HttpConnectionContext())

    // report successful binding

    binding.foreach { binding => println(s"gRPC server bound to: ${binding.localAddress}") }

    binding

    //#server

  }

}

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

客户端：

    val authenticator = AuthServiceClient(clientSettings)

    val futJwt = authenticator.getAuthToken(UCredential("",""))

    val jwt = Await.result(futJwt,.seconds).jwt

    println(s"got jwt: ${jwt}")

服务端：

class AuthServices() extends AuthServicePowerApi {

  //#service-request-reply

  override 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 = "(.*):(.*)".r

    val (host, port) = args() match {

      case hostPat(h, p) => (h, p)

      case _ => ("localhost", "")

    }

    // important to enable HTTP/2 in ActorSystem's config

    val 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.toClassic

    implicit val ec: ExecutionContext = system.executionContext

    val 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/8081

    val binding = Http().bindAndHandleAsync(

      serviceHandlers,

      interface = host,

      port = port,

      connectionContext = HttpConnectionContext())

    // report successful binding

    binding.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[_] = system

  private implicit val ec: ExecutionContext = system.executionContext

  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}"))

    }

  }

  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.Future

class AuthServices() extends AuthServicePowerApi {

  //#service-request-reply

  override 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.Try

import akka.Done

import akka.NotUsed

import akka.actor.ActorSystem

import akka.grpc.GrpcClientSettings

import akka.stream.scaladsl.Source

import learn.akka.grpc._

object AuthGreeterClient {

  def main(args: Array[String]): Unit = {

    implicit val sys = ActorSystem("grpc-client")

    implicit val ec = sys.dispatcher

    val clientSettings = GrpcClientSettings.fromConfig(GreeterService.name)

    val client = GreeterServiceClient(clientSettings)

    val authenticator = AuthServiceClient(clientSettings)

    val futJwt = authenticator.getAuthToken(UCredential("",""))

    val jwt = Await.result(futJwt,.seconds).jwt

    println(s"got jwt: ${jwt}")

    singleRequestReply(jwt)

    streamingRequest(jwt)

    streamingReply(jwt)

    streamingRequestReply(jwt)

    sys.scheduler.scheduleAtFixedRate(.second, .second)(new Runnable {

      override def run(): Unit = Try(singleRequestReply(jwt))

    })

    //#with-metadata

    def 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-metadata

    def 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, .minute)

    }

    def streamingRequestReply(jwt: String): Unit = {

      val requestStream: Source[HelloRequest, NotUsed] =

        Source

          .tick(.millis, .second, "tick")

          .zipWithIndex

          .map { case (_, i) => i }

          .map(i => HelloRequest(s"Alice-$i"))

          .take()

          .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, .minute)

    }

  }

}

巴特西

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

最新文章

热门文章