一. 摘要

此篇文章将用实例介绍grpc四种服务类型中的双向流式通信，模拟一个简单的聊天室功能。

二. 实践

整体项目如下：
image.png
其中cloud-grpc-java为maven项目，cloud-grpc-protos为定义接口项目。

1.通过protobuf定义接口和数据类型

在cloud-grpc-protos文件夹下创建doubleWayStream.proto,内容如下：

  1. syntax = "proto3";
  3. option go_package = "pbfs/double_way_stream";
  4. option java_multiple_files = true;
  5. option java_package = "com.cloud.grpc.doubleWayStream";
  6. option java_outer_classname = "DoubleWayStreamProto";
  7. option objc_class_prefix = "DWS";
  9. package double_way_stream;
  11. //双向流式
  12. service DoubleWayStreamService{
  13.   rpc DoubleWayStreamFun(stream RequestMessage) returns (stream ResponseMessage){}
  14. }
  16. message RequestMessage{
  17.   string req_msg = 1;
  18. }
  20. message ResponseMessage{
  21.   string rsp_msg = 1;
  22. }

以上，一个 简单 的双向流式RPC

2.maven 配置

创建一个如上图（cloud-grpc-java)的maven项目,pom.xml加入grpc开发相关配置，如下：

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  6.     <groupId>com.cloud.grpc</groupId>
  7.     <artifactId>java</artifactId>
  8.     <version>1.0.0.0</version>
  9.     <name>java</name>
  10.     <description>Demo project for grpc for java</description>
  12.     <properties>
  13.         <java.version>11</java.version>
  14.         <grpc.version>1.29.0</grpc.version>
  15.         <protobuf.version>3.11.0</protobuf.version>
  16.     </properties>
  18.     <dependencies>
  20.         <dependency>
  21.             <groupId>io.grpc</groupId>
  22.             <artifactId>grpc-netty-shaded</artifactId>
  23.             <version>${grpc.version}</version>
  24.         </dependency>
  26.         <dependency>
  27.             <groupId>io.grpc</groupId>
  28.             <artifactId>grpc-protobuf</artifactId>
  29.             <version>${grpc.version}</version>
  30.         </dependency>
  32.         <dependency>
  33.             <groupId>io.grpc</groupId>
  34.             <artifactId>grpc-stub</artifactId>
  35.             <version>${grpc.version}</version>
  36.         </dependency>
  38.         <dependency> <!-- necessary for Java 9+ -->
  39.             <groupId>org.apache.tomcat</groupId>
  40.             <artifactId>annotations-api</artifactId>
  41.             <version>6.0.53</version>
  42.             <scope>provided</scope>
  43.         </dependency>
  45.     </dependencies>
  47.     <build>
  49.         <extensions>
  50.             <extension>
  51.                 <groupId>kr.motd.maven</groupId>
  52.                 <artifactId>os-maven-plugin</artifactId>
  53.                 <version>1.5.0.Final</version>
  54.             </extension>
  55.         </extensions>
  57.         <plugins>
  58.             <plugin>
  59.                 <groupId>org.xolstice.maven.plugins</groupId>
  60.                 <artifactId>protobuf-maven-plugin</artifactId>
  61.                 <version>0.5.1</version>
  63.                 <configuration>
  64.                     <protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
  65.                    <pluginId>grpc-java</pluginId>
  66.                     <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
  67.                     <protoSourceRoot>../cloud-grpc-protos</protoSourceRoot>
  68.                 </configuration>
  70.                 <executions>
  71.                     <execution>
  72.                         <goals>
  73.                             <goal>compile</goal>
  74.                             <goal>compile-custom</goal>
  75.                         </goals>
  76.                     </execution>
  77.                 </executions>
  78.             </plugin>
  79.         </plugins>
  80.     </build>
  82. </project>

说明： ../cloud-grpc-protos 红色标注部分需要根据pom.xml与上面创建proto所在路径相匹配，这样protobuf 插件才会根据此目录找到定义的proto文件生成相关代码。

3.生成grpc,protobuf相关类

  1. mvn protobuf:compile
  2. mvn protobuf:compile-custom

同步pom.xml,即可在target下生成如下源码：
image.png

4.编写服务端，并运行

服务端源码如图：
image.png
首先实现proto文件中定义的rpc DoubleWayStream 的服务端流式响应接口，DoubleWayStreamIml.java如下

  1. package com.cloud.grpc.java.doubleWayStream.server;
  3. import com.cloud.grpc.doubleWayStream.DoubleWayStreamServiceGrpc;
  4. import com.cloud.grpc.doubleWayStream.RequestMessage;
  5. import com.cloud.grpc.doubleWayStream.ResponseMessage;
  6. import io.grpc.stub.StreamObserver;
  8. public class DoubleWayStreamIml extends DoubleWayStreamServiceGrpc.DoubleWayStreamServiceImplBase {
  10.     //声明此服务端流响应，为了后面通过控制台向后端发送消息
  11.     private StreamObserver<com.cloud.grpc.doubleWayStream.ResponseMessage> responseOb;
  13.     @Override
  14.     public StreamObserver<com.cloud.grpc.doubleWayStream.RequestMessage> doubleWayStreamFun(StreamObserver<com.cloud.grpc.doubleWayStream.ResponseMessage> responseObserver) {
  15.         this.responseOb=responseObserver;
  16.        return new StreamObserver<RequestMessage>() {
  17.             @Override
  18.             public void onNext(RequestMessage requestMessage) {
  19.                 System.out.println("[收到客户端消息]: " + requestMessage.getReqMsg());
  20.                 responseObserver.onNext(ResponseMessage.newBuilder().setRspMsg("hello client ,I'm Java grpc Server,your message '" + requestMessage.getReqMsg() + "'").build());
  21.             }
  23.             @Override
  24.             public void onError(Throwable throwable) {
  25.                 throwable.fillInStackTrace();
  26.             }
  28.             @Override
  29.             public void onCompleted() {
  30.                 responseObserver.onCompleted();
  31.             }
  32.         };
  34.     }
  35.     public StreamObserver<ResponseMessage> getResponseOb() {
  36.         return responseOb;
  37.     }
  38. }

如上图， private StreamObserver responseOb 声明此服务端流响应，为了后面通过控制台向后端发送消息。
编码服务端启动类DoubleWayStreamServer.java，如下

  1. package com.cloud.grpc.java.doubleWayStream.server;
  3. import com.cloud.grpc.doubleWayStream.ResponseMessage;
  4. import io.grpc.Server;
  5. import io.grpc.ServerBuilder;
  7. import java.io.IOException;
  8. import java.util.Scanner;
  10. public class DoubleWayStreamServer {
  12.     public static void main(String[] args) {
  13.         ServerBuilder<?> serverBuilder = ServerBuilder.forPort(8899);
  14.         DoubleWayStreamIml doubleWayStreamIml=new DoubleWayStreamIml();
  15.         serverBuilder.addService(doubleWayStreamIml);
  16.         Server server = serverBuilder.build();
  17.         try {
  18.             server.start();
  19.             //开启线程向客户端输入
  20.             new Thread(new Runnable() {
  21.                 @Override
  22.                 public void run() {
  23.                     Scanner scanner=new Scanner(System.in);
  24.                     for (;true;){
  25.                         String str=scanner.nextLine();
  26.                         if(str.equals("EOF")){
  27.                             break;
  28.                         }
  29.                         try {
  30.                             doubleWayStreamIml.getResponseOb().onNext(ResponseMessage.newBuilder().setRspMsg(str).build());
  31.                         }catch (Exception e){
  32.                             System.out.println("【异常】：没有客户端连接...");
  33.                             //一般客户端链接失败就会断开
  34.                             e.printStackTrace();
  35.                         }
  36.                   }
  37.                 }
  38.             }).start();
  39.             server.awaitTermination();
  41.         } catch (IOException | InterruptedException e) {
  42.             e.printStackTrace();
  43.         }
  45.     }
  46. }

如上图所示，开启一个新的线程，为了向客户端推送消息。
通过main函数启动，此时我们向控制台输入一条消息，结果如下：
image.png
可以看到，由于我们没有客户端与其建立连接，所以会在控制台打印：没有客户连接

5.编写，运行客户端

编写DoubleWayStreamClient.java,如下：

  1. package com.cloud.grpc.java.doubleWayStream.client;
  3. import com.cloud.grpc.doubleWayStream.DoubleWayStreamServiceGrpc;
  4. import com.cloud.grpc.doubleWayStream.RequestMessage;
  5. import com.cloud.grpc.doubleWayStream.ResponseMessage;
  6. import io.grpc.ManagedChannel;
  7. import io.grpc.ManagedChannelBuilder;
  8. import io.grpc.stub.StreamObserver;
  10. import java.util.Scanner;
  12. public class DoubleWayStreamClient {
  13.     public static void main(String[] args) {
  14.         //使用usePlaintext，否则使用加密连接
  15.         ManagedChannelBuilder<?> channelBuilder = ManagedChannelBuilder.forAddress("localhost", 8899).usePlaintext();
  17.         ManagedChannel channel = channelBuilder.build();
  19.         StreamObserver<RequestMessage> requestObserver = DoubleWayStreamServiceGrpc.newStub(channel).doubleWayStreamFun(new StreamObserver<ResponseMessage>() {
  20.             @Override
  21.             public void onNext(ResponseMessage value) {
  22.                 System.out.println("[收到服务端发来] : " + value.getRspMsg());
  23.             }
  25.             @Override
  26.             public void onError(Throwable t) {
  28.             }
  29.             @Override
  30.             public void onCompleted() {
  31.             }
  32.         });
  34.         Scanner scanner = new Scanner(System.in);
  35.         for (; true; ) {
  36.          String str=   scanner.nextLine();
  37.             if(str.equals("EOF")){
  38.                 requestObserver.onCompleted();
  39.                 break;
  40.             }
  41.             try {
  42.                 requestObserver.onNext(RequestMessage.newBuilder().setReqMsg(scanner.nextLine()).build());
  43.             } catch (Exception e) {
  44.                 e.printStackTrace();
  45.             }
  46.         }
  47.     }
  48. }

在上面源码中，我们实现了proto文件中rpc方法接口中的客户端流，在next方法中打印服务端的信息，执行main函数，在控制台输入”你好，我是客户端，我上线了”,
image.png
可见，服务端作出了响应，现在我们切换到服务端，让服务端想客户端发送消息，如下：
image.png
再看看客户端：
image.png
可见服务端可以向客户端发送消息。这样便实现了一个极其简单的聊天室。