java gRPC四种服务类型简单示例

一、gRPC 简介

gRPC 是Go实现的：一个高性能，开源，将移动和HTTP/2放在首位通用的RPC框架。使用gRPC可以在客户端调用不同机器上的服务端的方法，而客户端和服务端的开发语言和

运行环境可以有很多种，基本涵盖了主流语言和平台。双方交互的协议可以在proto文件中定义，客户端和服务端可以很方便的通过工具生成协议和代理代码。而消息的编码是采

用google protocol buffer，数据量小、速度快。

gRPC具有以下特点：

（1）基于 HTTP/2， 继而提供了连接多路复用、Body 和 Header 压缩等机制。可以节省带宽、降低TCP链接次数、节省CPU使用和延长电池寿命等。

（2）支持主流开发语言（C, C++, Python, PHP, Ruby, NodeJS, C#, Objective-C、Golang、Java）

（3）IDL (Interface Definition Language) 层使用了 Protocol Buffers, 非常适合团队的接口设计

二、编写示例工程

• 下载安装protobuf，过程省略

• 新建maven项目，过程省略，工程结构
  

• 完整的pom.xml

  <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  	<modelVersion>4.0.0</modelVersion>
  	<groupId>com.gm</groupId>
  	<artifactId>stream_grpc</artifactId>
  	<version>0.0.1-SNAPSHOT</version>
  
  	<name>stream_grpc</name>
  	<url>http://maven.apache.org</url>
  
  	<properties>
  		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  	</properties>
  
  
  	<dependencies>
  		<dependency>
  			<groupId>junit</groupId>
  			<artifactId>junit</artifactId>
  			<version>3.8.1</version>
  			<scope>test</scope>
  		</dependency>
  		<dependency>
  			<groupId>io.grpc</groupId>
  			<artifactId>grpc-all</artifactId>
  			<version>1.17.1</version>
  		</dependency>
  	</dependencies>
  
  
  	<build>
  		<extensions>
  			<extension>
  				<groupId>kr.motd.maven</groupId>
  				<artifactId>os-maven-plugin</artifactId>
  				<version>1.4.1.Final</version>
  			</extension>
  		</extensions>
  		<plugins>
  			<plugin>
  				<groupId>org.xolstice.maven.plugins</groupId>
  				<artifactId>protobuf-maven-plugin</artifactId>
  				<version>0.5.0</version>
  				<configuration>
  					<!-- The version of protoc must match protobuf-java. If you don't depend 
  						on protobuf-java directly, you will be transitively depending on the protobuf-java 
  						version that grpc depends on. -->
  					<protocArtifact>com.google.protobuf:protoc:3.0.0-beta-2:exe:${os.detected.classifier}</protocArtifact>
  					<pluginId>grpc-java</pluginId>
  					<pluginArtifact>io.grpc:protoc-gen-grpc-java:0.13.2:exe:${os.detected.classifier}</pluginArtifact>
  					<protocExecutable>protoc.exe</protocExecutable>
  				</configuration>
  				<executions>
  					<execution>
  						<goals>
  							<goal>compile</goal>
  							<goal>compile-custom</goal>
  						</goals>
  					</execution>
  				</executions>
  			</plugin>
  		</plugins>
  	</build>
  </project>
  

  此处需注意protoc的版本，我的为libprotoc 3.4.0，因此选择grpc-all 1.17.1版本

• 编写 proto 文件
  在src/main目录下新建proto目录，新建stream.proto文件，文件可名称任意

  syntax = "proto3";
  package stream;
  service StreamService {
    //简单rpc	
    rpc SimpleFun(RequestData) returns (ResponseData){}
    //服务器端流式rpc
    rpc ServerSideStreamFun(RequestData) returns (stream ResponseData){}
    //客户端流式rpc 
    rpc ClientSideStreamFun(stream RequestData) returns (ResponseData){}
    //双向流式rpc   
    rpc TwoWayStreamFun(stream RequestData) returns (stream ResponseData){}
  }
  message RequestData {
    string text = 1;
  }
  
  message ResponseData {
    string text = 1;
  }
  

• 编译 protobuf
  执行命令：mvn generate-sources在target目下生成下图的文件
  

• 文件复制
  在src/main/java目录下新建stream包，具体包取决于proto文件中的package名称 ，并java文件复制到新建包中

三、服务类型示例

gRPC存在四种服务类型：

简单rpc

就是一般的rpc调用，一个请求对象对应一个返回对象

• proto语法，参考stream.proto文件：

  //简单rpc	
  rpc SimpleFun(RequestData) returns (ResponseData){}
  

• 服务实现

  @Override
  public void simpleFun(RequestData request, StreamObserver<ResponseData> responseObserver) {
  	// TODO Auto-generated method stub
  	System.out.println("请求参数：" + request.getText());
  	responseObserver.onNext(ResponseData.newBuilder().setText("hello gRPC").build());
  	responseObserver.onCompleted();
  }
  

• server实现，后三种服务类型server实现相同

  package simple.fun;
  
  import java.io.IOException;
  
  import stream.ServiceImpl;
  import stream.StreamServiceGrpc;
  import stream.StreamServiceGrpc.StreamService;
  
  public class Server {
  	private static int port = 8883;
  	private static io.grpc.Server server;
  
  	public void run() {
  		StreamService serviceImpl = new ServiceImpl();
  		server = io.grpc.ServerBuilder.forPort(port).addService(StreamServiceGrpc.bindService(serviceImpl)).build();
  		try {
  			server.start();
  			System.out.println("Server start success on port:" + port);
  			server.awaitTermination();
  		} catch (IOException e) {
  			e.printStackTrace();
  		} catch (InterruptedException e) {
  			e.printStackTrace();
  		}
  	}
  
  	public static void main(String[] args) {
  		Server server = new Server();
  		server.run();
  	}
  }
  

• client实现

  package simple.fun;
  
  import java.io.FileNotFoundException;
  import java.io.IOException;
  
  import io.grpc.ManagedChannel;
  import io.grpc.ManagedChannelBuilder;
  import stream.Stream.RequestData;
  import stream.Stream.ResponseData;
  import stream.StreamServiceGrpc;
  import stream.StreamServiceGrpc.StreamServiceBlockingStub;
  
  public class Client {
  
  	public static void main(String[] args) throws FileNotFoundException, IOException {	
  		ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 8883).usePlaintext(true).build();
  		StreamServiceBlockingStub stub = StreamServiceGrpc.newBlockingStub(channel);
  		
  		long start = System.currentTimeMillis();
  		for (int i = 0; i < 10; i++) {
  			RequestData requestData = RequestData.newBuilder().setText("hello world").build();
  			ResponseData responseData = stub.simpleFun(requestData);
  			System.out.println(responseData.getText());
  		}
  		System.out.println(System.currentTimeMillis() - start + " MS");
  		channel.shutdown();
  	}
  }
  
  

• 执行结果
  

服务端流式rpc

一个请求对象，服务端可以传回多个结果对象

• proto语法，参考stream.proto文件：

  //服务器端流式rpc
  rpc ServerSideStreamFun(RequestData) returns (stream ResponseData){}
  

• 服务实现

  @Override
  public void serverSideStreamFun(RequestData request, StreamObserver<ResponseData> responseObserver) {
  	// TODO Auto-generated method stub
  	System.out.println("请求参数：" + request.getText());
  	for (int i = 0; i < 10; i++) {
  		responseObserver.onNext(ResponseData.newBuilder().setText("你好" + i).build());
  	}
  	responseObserver.onCompleted();
  }
  

• client实现

  package server.side.stream;
  
  import java.io.FileNotFoundException;
  import java.io.IOException;
  import java.util.Iterator;
  
  import io.grpc.ManagedChannel;
  import io.grpc.ManagedChannelBuilder;
  import stream.Stream.RequestData;
  import stream.Stream.ResponseData;
  import stream.StreamServiceGrpc;
  import stream.StreamServiceGrpc.StreamServiceBlockingStub;
  
  public class Client {
  
  	public static void main(String[] args) throws FileNotFoundException, IOException {
  		RequestData requestData = RequestData.newBuilder().setText("hello world").build();
  		ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 8883).usePlaintext(true).build();
  		StreamServiceBlockingStub stub = StreamServiceGrpc.newBlockingStub(channel);
  
  		Iterator<ResponseData> it = stub.serverSideStreamFun(requestData);
  		long start = System.currentTimeMillis();
  		while (it.hasNext()) {
  			System.out.println(it.next().getText());
  		}
  		channel.shutdown();
  		System.out.println(System.currentTimeMillis() - start + " MS");
  
  	}
  }
  
  

• 执行结果
  -

客户端流式rpc

客户端传入多个请求对象，服务端返回一个响应结果

• proto语法，参考stream.proto文件：

  //客户端流式rpc 
  rpc ClientSideStreamFun(stream RequestData) returns (ResponseData){}
  

• 服务实现

  @Override
  public StreamObserver<RequestData> clientSideStreamFun(StreamObserver<ResponseData> responseObserver) {
  	// TODO Auto-generated method stub
  	return new StreamObserver<RequestData>() {
  		private ResponseData.Builder builder = ResponseData.newBuilder();
  
  		@Override
  		public void onNext(RequestData value) {
  			// TODO Auto-generated method stub
  			System.out.println("请求参数：" + value.getText());
  
  		}
  
  		@Override
  		public void onError(Throwable t) {
  			// TODO Auto-generated method stub
  
  		}
  
  		@Override
  		public void onCompleted() {
  			// TODO Auto-generated method stub
  			builder.setText("数据接收完成");
  			responseObserver.onNext(builder.build());
  			responseObserver.onCompleted();
  		}
  
  	};
  }
  

• client实现

  package client.side.stream;
  
  import java.io.FileNotFoundException;
  import java.io.IOException;
  import java.util.Iterator;
  
  import io.grpc.ManagedChannel;
  import io.grpc.ManagedChannelBuilder;
  import io.grpc.stub.StreamObserver;
  import stream.Stream.RequestData;
  import stream.Stream.ResponseData;
  import stream.StreamServiceGrpc;
  import stream.StreamServiceGrpc.StreamServiceBlockingStub;
  import stream.StreamServiceGrpc.StreamServiceStub;
  
  public class Client {
  
  	public static void main(String[] args) throws FileNotFoundException, IOException, InterruptedException {
  		ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 8883).usePlaintext(true).build();
  		StreamServiceStub asyncStub = StreamServiceGrpc.newStub(channel);
  
  		StreamObserver<ResponseData> responseData = new StreamObserver<ResponseData>() {
  			@Override
  			public void onNext(ResponseData value) {
  				// TODO Auto-generated method stub
  				System.out.println(value.getText());
  			}
  
  			@Override
  			public void onError(Throwable t) {
  				// TODO Auto-generated method stub
  				t.printStackTrace();
  			}
  
  			@Override
  			public void onCompleted() {
  				// TODO Auto-generated method stub
  				// 关闭channel
  				channel.shutdown();
  			}
  		};
  
  		StreamObserver<RequestData> requestData = asyncStub.clientSideStreamFun(responseData);
  		long start = System.currentTimeMillis();
  		for (int i = 0; i < 10; i++) {
  			requestData.onNext(RequestData.newBuilder().setText("你好" + i).build());
  		}
  		requestData.onCompleted();
  		System.out.println(System.currentTimeMillis() - start + " MS");
  		// 由于是异步获得结果，所以sleep 10秒
  		Thread.sleep(10000);
  	}
  }
  

• 执行结果
  

双向流式rpc

结合客户端流式rpc和服务端流式rpc，可以传入多个对象，返回多个响应对象

• proto语法，参考stream.proto文件：

  //双向流式rpc   
  rpc TwoWayStreamFun(stream RequestData) returns (stream ResponseData){}
  

• 服务实现

  	@Override
  public StreamObserver<RequestData> twoWayStreamFun(StreamObserver<ResponseData> responseObserver) {
  	// TODO Auto-generated method stub
  	return new StreamObserver<RequestData>() {
  
  		@Override
  		public void onNext(RequestData value) {
  			// TODO Auto-generated method stub
  			System.out.println("请求参数：" + value.getText());
  			responseObserver.onNext(ResponseData.newBuilder().setText(value.getText() + "，欢迎你的加入").build());
  		}
  
  		@Override
  		public void onError(Throwable t) {
  			// TODO Auto-generated method stub
  			t.printStackTrace();
  		}
  
  		@Override
  		public void onCompleted() {
  			// TODO Auto-generated method stub
  			responseObserver.onCompleted();
  		}
  
  	};
  }
  

• client实现

  package two.way.stream;
  
  import java.io.FileNotFoundException;
  import java.io.IOException;
  import java.util.Iterator;
  
  import io.grpc.ManagedChannel;
  import io.grpc.ManagedChannelBuilder;
  import io.grpc.stub.StreamObserver;
  import stream.Stream.RequestData;
  import stream.Stream.ResponseData;
  import stream.StreamServiceGrpc;
  import stream.StreamServiceGrpc.StreamServiceBlockingStub;
  import stream.StreamServiceGrpc.StreamServiceStub;
  
  public class Client {
  
  	public static void main(String[] args) throws FileNotFoundException, IOException, InterruptedException {
  		ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 8883).usePlaintext(true).build();
  		StreamServiceStub asyncStub = StreamServiceGrpc.newStub(channel);
  
  		StreamObserver<ResponseData> responseData = new StreamObserver<ResponseData>() {
  			@Override
  			public void onNext(ResponseData value) {
  				// TODO Auto-generated method stub
  				System.out.println(value.getText());
  			}
  
  			@Override
  			public void onError(Throwable t) {
  				// TODO Auto-generated method stub
  				t.printStackTrace();
  			}
  
  			@Override
  			public void onCompleted() {
  				// TODO Auto-generated method stub
  				// 关闭channel
  				channel.shutdown();
  			}
  		};
  
  		StreamObserver<RequestData> requestData = asyncStub.twoWayStreamFun(responseData);
  		long start = System.currentTimeMillis();
  		for (int i = 0; i < 10; i++) {
  			requestData.onNext(RequestData.newBuilder().setText("你好" + i).build());
  		}
  		requestData.onCompleted();
  		System.out.println(System.currentTimeMillis() - start + " MS");
  		// 由于是异步获得结果，所以sleep 10秒
  		Thread.sleep(10000);
  	}
  }
  
  

• 执行结果
  
• 完整代码
  https://github.com/gm19900510/stream_grpc_java