了解proto3后,接下來看官方Demo作為訓(xùn)練,這里建議看一遍之后自己動(dòng)手搭建出來,一方面鞏固之前的知識(shí),一方面是對(duì)整個(gè)流程更加熟悉.

官方Demo地址: https://github.com/grpc/grpc-java
例子是一個(gè)簡(jiǎn)單的路由映射的應(yīng)用，它允許客戶端獲取路由特性的信息，生成路由的總結(jié)，以及交互路由信息，如服務(wù)器和其他客戶端的流量更新.

1.1定義服務(wù)

也就是寫proto文件

//指定proto3格式 syntax = "proto3"; //一些生成代碼的設(shè)置 option java_multiple_files = true;//以外部類模式生成 option java_package = "cn.mrdear.route";//所在包名 option java_outer_classname = "RouteProto";//最外層類名稱//定義服務(wù) service RouteGuide{//得到指定點(diǎn)的feature//一個(gè) 簡(jiǎn)單 RPC ， 客戶端使用存根發(fā)送請(qǐng)求到服務(wù)器并等待響應(yīng)返回，就像平常的函數(shù)調(diào)用一樣。rpc GetFeature(Point) returns (Feature) {}//獲取一個(gè)矩形內(nèi)的點(diǎn)//一個(gè) 服務(wù)器端流式 RPC ， 客戶端發(fā)送請(qǐng)求到服務(wù)器，拿到一個(gè)流去讀取返回的消息序列。 客戶端讀取返回的流，//直到里面沒有任何消息。從例子中可以看出，通過在 響應(yīng) 類型前插入 stream 關(guān)鍵字，可以指定一個(gè)服務(wù)器端的流方法。rpc ListFeatures(Rectangle) returns (stream Feature){}//記錄該點(diǎn)//一個(gè) 客戶端流式 RPC ， 客戶端寫入一個(gè)消息序列并將其發(fā)送到服務(wù)器，同樣也是使用流。一旦客戶端完成寫入消息，//它等待服務(wù)器完成讀取返回它的響應(yīng)。通過在 請(qǐng)求 類型前指定 stream 關(guān)鍵字來指定一個(gè)客戶端的流方法。rpc RecordRoute(stream Point) returns (RouteSummary){}//路由交流//一個(gè) 雙向流式 RPC 是雙方使用讀寫流去發(fā)送一個(gè)消息序列。兩個(gè)流獨(dú)立操作，因此客戶端和服務(wù)器//可以以任意喜歡的順序讀寫：比如， 服務(wù)器可以在寫入響應(yīng)前等待接收所有的客戶端消息，或者可以交替 的讀取和寫入消息，//或者其他讀寫的組合。每個(gè)流中的消息順序被預(yù)留。你可以通過在請(qǐng)求和響應(yīng)前加 stream 關(guān)鍵字去制定方法的類型。rpc RouteChat(stream RouteNote) returns (stream RouteNote){} }//代表經(jīng)緯度 message Point {int32 latitude = 1;int32 longitude = 2; } //由兩個(gè)點(diǎn)確定的一個(gè)方塊 message Rectangle{Point lo = 1;Point hi = 2; } //某一位置的名稱 message Feature {string name = 1;Point location = 2; }// Not used in the RPC. Instead, this is here for the form serialized to disk. message FeatureDatabase {repeated Feature feature = 1; } //給某一點(diǎn)發(fā)送消息 message RouteNote{Point location = 1;string message = 2; }//記錄收到的信息 message RouteSummary{int32 point_count = 1;int32 feture_count = 2;int32 distance = 3;int32 elapsed_time = 4; }

執(zhí)行mvn compile生成如下代碼:

Paste_Image.png

1.2編寫RouteGuideService

該類就是這個(gè)項(xiàng)目所提供給外部的功能.該類需要繼承RouteGuideGrpc.RouteGuideImplBase,這個(gè)類提供了我們所定義分服務(wù)接口,繼承后覆蓋需要實(shí)現(xiàn)的自定義方法.

簡(jiǎn)單 RPC
簡(jiǎn)單RPC和普通方法調(diào)用形式差不多,客戶端傳來一個(gè)實(shí)體,服務(wù)端返回一個(gè)實(shí)體.

@Overridepublic void getFeature(Point request, StreamObserver<Feature> responseObserver) {System.out.println("getFeature得到的請(qǐng)求參數(shù): " + request.toString()); // responseObserver.onError(); 代表請(qǐng)求出錯(cuò)responseObserver.onNext(checkFeature(request));//包裝返回信息responseObserver.onCompleted();//結(jié)束一次請(qǐng)求}//找到復(fù)核的featureprivate Feature checkFeature(Point location) {for (Feature feature : features) {if (feature.getLocation().getLatitude() == location.getLatitude()&& feature.getLocation().getLongitude() == location.getLongitude()) {return feature;}}// No feature was found, return an unnamed feature.return Feature.newBuilder().setName("").setLocation(location).build();}

其中StreamObserver<Feature>是一個(gè)應(yīng)答觀察者,用于封裝返回的信息,服務(wù)器把該信息傳給客戶端.請(qǐng)求結(jié)束要調(diào)用onCompleted()方法.

服務(wù)器端流式 RPC
在proto文件中聲明了stream,但是從接口上看不出來和簡(jiǎn)單RPC的區(qū)別,代碼中最主要的區(qū)別是多次調(diào)用responseObserver.onNext()的方法,最后完成時(shí)寫回?cái)?shù)據(jù).

@Overridepublic void listFeatures(Rectangle request, StreamObserver<Feature> responseObserver) {int left = min(request.getLo().getLongitude(), request.getHi().getLongitude());int right = max(request.getLo().getLongitude(), request.getHi().getLongitude());int top = max(request.getLo().getLatitude(), request.getHi().getLatitude());int bottom = min(request.getLo().getLatitude(), request.getHi().getLatitude());for (Feature feature : features) {//如果不存在則繼續(xù)if (!RouteGuideUtil.exists(feature)) {continue;}int lat = feature.getLocation().getLatitude();int lon = feature.getLocation().getLongitude();if (lon >= left && lon <= right && lat >= bottom && lat <= top) {//找到符合的就寫入responseObserver.onNext(feature);}}//最后標(biāo)識(shí)完成responseObserver.onCompleted();}

客戶端流式 RPC
服務(wù)端就需要一直監(jiān)控客戶端寫入情況,因此需要一個(gè)StreamObserver接口,其中onNext方法會(huì)在客戶端每次寫入時(shí)調(diào)用,當(dāng)寫入完畢時(shí)調(diào)用onCompleted()方法.具體還要到后面客戶端調(diào)用分析.

@Overridepublic StreamObserver<Point> recordRoute(StreamObserver<RouteSummary> responseObserver) {return new StreamObserver<Point>() {int pointCount;int featureCount;int distance;Point previous;long startTime = System.nanoTime();//客戶端每寫入一個(gè)Point,服務(wù)端就會(huì)調(diào)用該方法@Overridepublic void onNext(Point point) {System.out.println("recordRoute得到的請(qǐng)求參數(shù): " + point.toString());pointCount++;if (RouteGuideUtil.exists(checkFeature(point))) {featureCount++;}if (previous != null) {distance += calcDistance(previous, point);}previous = point;}@Overridepublic void onError(Throwable throwable) {throwable.printStackTrace();System.err.println("Encountered error in recordRoute");}//客戶端寫入結(jié)束時(shí)調(diào)用@Overridepublic void onCompleted() {long seconds = NANOSECONDS.toSeconds(System.nanoTime() - startTime);responseObserver.onNext(RouteSummary.newBuilder().setPointCount(pointCount).setFetureCount(featureCount).setDistance(distance).setElapsedTime((int) seconds).build());responseObserver.onCompleted();}};}

雙向流式 RPC
和客戶端流式RPC差不多.

@Overridepublic StreamObserver<RouteNote> routeChat(StreamObserver<RouteNote> responseObserver) {return new StreamObserver<RouteNote>() {@Overridepublic void onNext(RouteNote note) {List<RouteNote> notes = getOrCreateNotes(note.getLocation());for (RouteNote prevNote : notes.toArray(new RouteNote[0])) {responseObserver.onNext(prevNote);}notes.add(note);}@Overridepublic void onError(Throwable t) {t.printStackTrace();System.err.println("Encountered error in routeChat");}@Overridepublic void onCompleted() {responseObserver.onCompleted();}};}

1.3創(chuàng)建服務(wù)端

和Helloworld一樣的形式,最主要的是addService(new RouteGuideService(features)),這里把需要注冊(cè)的服務(wù)給注冊(cè)上.

public class RouteGuideServer {private final int port;//服務(wù)端端口private final Server server;//服務(wù)器public RouteGuideServer(int port) throws IOException {this.port = port;//獲取初始化數(shù)據(jù)List<Feature> features = RouteGuideUtil.parseFeatures(RouteGuideUtil.getDefaultFeaturesFile());//初始化Server參數(shù)server = ServerBuilder.forPort(port)//添加指定服務(wù).addService(new RouteGuideService(features)).build();}/*** 啟動(dòng)服務(wù)*/public void start() throws IOException {server.start();System.out.println("Server started, listening on " + port);//程序退出時(shí)關(guān)閉資源Runtime.getRuntime().addShutdownHook(new Thread(() -> {System.err.println("*** shutting down gRPC server since JVM is shutting down");RouteGuideServer.this.stop();System.err.println("*** server shut down");}));}/*** 關(guān)閉服務(wù)*/public void stop() {if (server != null) {server.shutdown();}}/*** 使得server一直處于運(yùn)行狀態(tài)*/private void blockUntilShutdown() throws InterruptedException {if (server != null) {server.awaitTermination();}}public static void main(String[] args) throws IOException, InterruptedException {RouteGuideServer server = new RouteGuideServer(50051);server.start();server.blockUntilShutdown();}}

1.4編寫客戶端

客戶端需要一個(gè)channel和一個(gè)存根blockingStub或者asyncStub根據(jù)業(yè)務(wù)需要選擇同步或者異步.

private final ManagedChannel channel;//grpc信道,需要指定端口和地址private final RouteGuideGrpc.RouteGuideBlockingStub blockingStub;//阻塞/同步存根private final RouteGuideGrpc.RouteGuideStub asyncStub;//非阻塞,異步存根public RouteGuideClient(String host,int port) {//創(chuàng)建信道channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext(true).build();//創(chuàng)建存根blockingStub = RouteGuideGrpc.newBlockingStub(channel);asyncStub = RouteGuideGrpc.newStub(channel);}/*** 關(guān)閉方法*/public void shutdown() throws InterruptedException {channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);}

簡(jiǎn)單grpc
和調(diào)用普通方法形式差不多.

public void getFeature(int lat,int lon){System.out.println("start getFeature");Point request = Point.newBuilder().setLatitude(lat).setLongitude(lon).build();Feature feature;try {//同步阻塞調(diào)用feature = blockingStub.getFeature(request);System.out.println("getFeature服務(wù)端返回 :" + feature);} catch (StatusRuntimeException e) {System.out.println("RPC failed " +e.getStatus());}}

調(diào)用代碼:

public static void main(String[] args) throws InterruptedException {RouteGuideClient client = new RouteGuideClient("localhost", 50051);try {client.getFeature(409146138, -746188906);//成功案例client.getFeature(0, 0);//失敗案例} finally {client.shutdown();}}

客戶端日志

Paste_Image.png

服務(wù)端日志(參數(shù)都為0的時(shí)候,這邊并沒拿到參數(shù))

Paste_Image.png

服務(wù)器端流式 RPC
和簡(jiǎn)單RPC差不多,只不過返回的是一個(gè)集合類.

//2.服務(wù)端流式RPCpublic void listFeatures(int lowLat, int lowLon, int hiLat, int hiLon){System.out.println("start listFeatures");Rectangle request =Rectangle.newBuilder().setLo(Point.newBuilder().setLatitude(lowLat).setLongitude(lowLon).build()).setHi(Point.newBuilder().setLatitude(hiLat).setLongitude(hiLon).build()).build();Iterator<Feature> features;try {features = blockingStub.listFeatures(request);for (int i = 1; features.hasNext(); i++) {Feature feature = features.next();System.out.println("getFeature服務(wù)端返回 :" + feature);}} catch (Exception e) {System.out.println("RPC failed " +e.getMessage());}}

客戶端日志:

Paste_Image.png

服務(wù)端日志:

Paste_Image.png

客戶端流式 RPC
該種方式兩遍都是異步操作,所以需要互相監(jiān)聽,也因此需要使用阻塞存根.服務(wù)端監(jiān)聽Point的寫入,客戶端監(jiān)聽RouteSummary的寫回.

public void recordRoute(List<Feature> features, int numPoints) throws InterruptedException {System.out.println("start recordRoute");final CountDownLatch finishLatch = new CountDownLatch(1);//建一個(gè)應(yīng)答者接受返回?cái)?shù)據(jù)StreamObserver<RouteSummary> responseObserver = new StreamObserver<RouteSummary>() {@Overridepublic void onNext(RouteSummary summary) {System.out.println("recordRoute服務(wù)端返回 :" + summary);}@Overridepublic void onError(Throwable t) {System.out.println("RecordRoute Failed");finishLatch.countDown();}@Overridepublic void onCompleted() {System.out.println("RecordRoute finish");finishLatch.countDown();}};//客戶端寫入操作StreamObserver<Point> requestObserver = asyncStub.recordRoute(responseObserver);Random random = new Random();try {for (int i = 0; i < numPoints; ++i) {int index = random.nextInt(features.size());Point point = features.get(index).getLocation();System.out.println("客戶端寫入point:" + point);requestObserver.onNext(point);Thread.sleep(random.nextInt(1000) + 500);if (finishLatch.getCount() == 0) {return;}}} catch (RuntimeException e) {requestObserver.onError(e);throw e;}//標(biāo)識(shí)已經(jīng)寫完requestObserver.onCompleted();// Receiving happens asynchronouslyif (!finishLatch.await(1, TimeUnit.MINUTES)) {System.out.println("recordRoute can not finish within 1 minutes");}}

客戶端日志:

Paste_Image.png

服務(wù)端日志:

Paste_Image.png

雙向流式 RPC
和客戶端流式RPC比較接近,同樣都需要雙方監(jiān)控.

public CountDownLatch routeChat() {System.out.println("start routeChat");final CountDownLatch finishLatch = new CountDownLatch(1);//寫入監(jiān)聽StreamObserver<RouteNote> requestObserver =//寫回監(jiān)聽asyncStub.routeChat(new StreamObserver<RouteNote>() {//服務(wù)端每寫回一個(gè)操作就調(diào)用@Overridepublic void onNext(RouteNote note) {System.out.println("服務(wù)端寫回: " + note);}@Overridepublic void onError(Throwable t) {t.printStackTrace();System.out.println("RouteChat Failed:");finishLatch.countDown();}@Overridepublic void onCompleted() {System.out.println("Finished RouteChat");finishLatch.countDown();}});try {RouteNote[] requests ={newNote("First message", 0, 0), newNote("Second message", 0, 1),newNote("Third message", 1, 0), newNote("Fourth message", 1, 1)};for (RouteNote request : requests) {System.out.println("客戶端寫入:" + request);requestObserver.onNext(request);}} catch (RuntimeException e) {requestObserver.onError(e);throw e;}//標(biāo)識(shí)寫完requestObserver.onCompleted();return finishLatch;}

這里調(diào)用需要特殊處理下;

CountDownLatch finishLatch = client.routeChat();if (!finishLatch.await(1, TimeUnit.MINUTES)) {System.out.println("routeChat can not finish within 1 minutes");}

客戶端日志:

Paste_Image.png

服務(wù)端日志:

Paste_Image.png

官方Demo之后,入門算結(jié)束,接下來就要看詳細(xì)的官方文檔,然后在項(xiàng)目中使用,這個(gè)過程會(huì)遇到不少問題,解決這些問題就是對(duì)這個(gè)技術(shù)的熟練.

附錄:

相關(guān)代碼: https://github.com/nl101531/JavaWEB

作者：此博廢棄_更新在個(gè)人博客
鏈接：https://www.jianshu.com/p/39c9eedba2c2
來源：簡(jiǎn)書
著作權(quán)歸作者所有。商業(yè)轉(zhuǎn)載請(qǐng)聯(lián)系作者獲得授權(quán)，非商業(yè)轉(zhuǎn)載請(qǐng)注明出處。

總結(jié)

以上是生活随笔為你收集整理的gRPC学习记录(四)--官方Demo的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。