PICE（6）：叢集環境裡多異類端點gRPC Streaming - Heterogeneous multi-endpoints gRPC streaming
RPC/">gRPC Streaming的操作物件由服務端和客戶端組成。在一個包含了多個不同服務的叢集環境中可能需要從一個服務裡呼叫另一個服務端提供的服務。這時呼叫服務端又成為了提供服務端的客戶端了（服務消費端）。那麼如果我們用streaming形式來提交服務需求及獲取計算結果就是以一個服務端為Source另一個服務端為通過式passthrough Flow的stream運算了。講詳細點就是請求方用需求構建Source，以連線Flow的方式把需求傳遞給服務提供方。服務提供方在Flow內部對需求進行處理後再把結果返回來，請求方run這個連線的stream應該就可以得到需要的結果了。下面我們就針對以上場景在一個由JDBC,Cassandra,MongoDB幾種gRPC服務組成的叢集環境裡示範在這幾個服務之間的stream連線和運算。
首先，我們設計一個簡單但比較有代表性的例子：從JDBC的客戶端傳一個字元型訊息hello給JDBC服務端、JDBC服務端在hello後面新增“,from jdbc to cassandra”然後通過Cassandra客戶端把訊息當作請求傳給Cassandra服務端、Cassandra服務端在訊息後面再加上“,from cassandra to mongo”並通過MongoDB客戶端把訊息傳給MongoDB服務端、最後MongoDB服務端在訊息後面新增“,mongo says hi”。整個stream的形狀是 jdbc-client->jdbc-service->cassandra-service-mongodb-service。如果run這個stream得到的結果應該是一個描述完整移動路徑的訊息。從請求-服務角度來描述：我們可以把每個節點訊息更新處理當作某種完整的資料處理過程。
以下分別是JDBC,Cassandra,MongoDB gRPC IDL定義：
service JDBCServices {
rpc greeting(stream HelloMsg) returns (stream HelloMsg) {}
}
service CQLServices {
rpc greeting(stream HelloMsg) returns (stream HelloMsg) {}
}
service MGOServices {
rpc greeting(stream HelloMsg) returns (stream HelloMsg) {}
}
三個服務共用了protobuf訊息型別HelloMsg。我們把共用的訊息統一放到一個common.proto檔案裡：
syntax = "proto3";
package sdp.grpc.services;
message HelloMsg {
string hello = 1;
}
message DataRow {
string countyname = 1;
string statename = 2;
int32 reportyear = 3;
int32 value = 4;
}
然後在示範應用的.proto檔案中用import 把所有protobuf,gRPC服務定義都集中起來：
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 package
flat_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"
};
/*
* Demoes various customization options provided by ScalaPBs.
*/
package sdp.grpc.services;
import "misc/sdp.proto";
import "common.proto";
import "cql/cql.proto";
import "jdbc/jdbc.proto";
import "mgo/mgo.proto";
下面我們把最核心的服務實現挑出來講解一下，先看看Cassandra服務的實現：
import sdp.grpc.mongo.client.MGOClient
class CQLStreamingServices(implicit ec: ExecutionContextExecutor,
mat: ActorMaterializer,session: Session)
extends CqlGrpcAkkaStream.CQLServices with LogSupport{
val mongoClient = new MGOClient
val stub = mongoClient.stub
def sayHelloTo(msg: String): Flow[HelloMsg, HelloMsg, NotUsed] =
Flow[HelloMsg].map { r => HelloMsg(r.hello + msg)}
.via(stub.greeting)
override def greeting: Flow[HelloMsg, HelloMsg, NotUsed] =
Flow[HelloMsg]
.via(sayHelloTo(",from cassandra to mongo"))
}
streaming方式的gRPC服務其實就是一個akka-stream的Flow[R1,R2,M]，它把收到的資料R1處理後轉換成R2輸出。在處理R1的環節裡可能會需要其它服務的運算結果。在以上例子裡CQLService把收到的訊息加工轉換後傳給MGOService並等待MGOService再深度加工返還的結果，所以sayHelloTo還是一個有兩個節點的Flow：在第一個節點中對收到的訊息進行加工，第二個節點把加工的訊息傳給另一個服務並連線它的運算結果作為本身最終的輸出。呼叫其它跨叢集節點的服務必須經該服務的gRPC客戶端進行，這裡呼叫的MGOClient：
package sdp.grpc.mongo.client
import sdp.grpc.services._
import sdp.logging.LogSupport
import io.grpc._
import common._
import sdp.grpc.services._
import akka.stream.scaladsl._
import akka.NotUsed
class MGOClient extends LogSupport {
val channel = ManagedChannelBuilder
.forAddress("localhost", 50051)
.usePlaintext()
.build()
val stub = MgoGrpcAkkaStream.stub(channel)
}
JDBCService連線CQLService, CQLService連線MGOService：
import sdp.grpc.cassandra.client.CQLClient
class JDBCStreamingServices(implicit ec: ExecutionContextExecutor)
extends JdbcGrpcAkkaStream.JDBCServices with LogSupport {
val cassandraClient = new CQLClient
val stub = cassandraClient.stub
def sayHelloTo(msg: String): Flow[HelloMsg,HelloMsg,NotUsed] =
Flow[HelloMsg]
.map {r => HelloMsg(r.hello + msg)}
.via(stub.greeting)
override def greeting: Flow[HelloMsg, HelloMsg, NotUsed] =
Flow[HelloMsg]
.via(sayHelloTo(",from jdbc to cassandra"))
}
最後我們用DemoApp來示範整個過程：
package demo.sdp.grpc
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ThrottleMode}
import sdp.grpc.jdbc.client.JDBCClient
object DemoApp extends App {
implicit val system = ActorSystem("jdbcClient")
implicit val mat = ActorMaterializer.create(system)
implicit val ec = system.dispatcher
val jdbcClient = new JDBCClient
jdbcClient.sayHello.runForeach(r => println(r.hello))
scala.io.StdIn.readLine()
mat.shutdown()
system.terminate()
}
DemoApp呼叫了JDBCClient：
package sdp.grpc.jdbc.client
import sdp.grpc.services._
import sdp.logging.LogSupport
import io.grpc._
import common._
import sdp.grpc.services._
import akka.stream.scaladsl._
import akka.NotUsed
class JDBCClient extends LogSupport {
val channel = ManagedChannelBuilder
.forAddress("localhost", 50053)
.usePlaintext()
.build()
val stub = JdbcGrpcAkkaStream.stub(channel)
def sayHello: Source[HelloMsg, NotUsed] = {
val row = HelloMsg("hello ")
val rows = List.fill[HelloMsg](100)(row)
Source
.fromIterator(() => rows.iterator)
.via(stub.greeting)
}
}
執行DemoApp顯示的結果：
hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi ...