1. GRPC
从上述可知 grpc一共有四种通信模型，如下所示，接下来就来仔细探讨一下这四种模型
image-20210531165018938
1.1 Unary
这种通信模型也是最简单的模型，客户端发送单一请求消息，服务端回复一个单一响应
接下来实现一个经典案例，客户端向服务端发送字符串，服务端返回该字符串和uuid，如下图
image-20210604140031013
实现起来也较为简单。具体步骤如下：
- 构建环境
- 构建服务端
- 构建客户端
- 测试消息发送
接下来就挨个实现步骤
1.1.1 common
创建一个maven项目(grpc-parent)，打包方式为pom,其pom.xml内容如下：
<?xml version="1.0" encoding="UTF-8"?>
<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.tomato.wangzh</groupId>
<artifactId>grpc-parent</artifactId>
<packaging>pom</packaging>
<version>1.0-SNAPSHOT</version>
<modules>
<module>grpc-common</module>
<module>grpc-server</module>
<module>grpc-client</module>
</modules>
<properties>
<grpc.version>1.36.0</grpc.version>
<protobuf.version>3.3.0</protobuf.version>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<os-maven.plugin.version>1.5.0.Final</os-maven.plugin.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
<version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
<version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty</artifactId>
<version>${grpc.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
</project>
创建maven模块(grpc-common),其pom.xml如下
<?xml version="1.0" encoding="UTF-8"?>
<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">
<parent>
<artifactId>grpc-parent</artifactId>
<groupId>com.tomato.wangzh</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>grpc-common</artifactId>
<dependencies>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
</dependency>
</dependencies>
<build>
<extensions>
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>${os-maven.plugin.version}</version>
</extension>
</extensions>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
<version>0.5.0</version>
<configuration>
<protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
<pluginId>grpc-java</pluginId>
<pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}
</pluginArtifact>
<protoSourceRoot>src/main/resources/proto</protoSourceRoot>
</configuration>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>compile-custom</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
在grpc-common/src/main/resources下，创建proto/GrpcHello.proto，内容如下：
syntax = "proto3";
package tomato;
option java_package = "com.tomato.wangzh.grpc.common";
option java_multiple_files = true;
// 定义请求消息
message HelloRequest {
string param = 1;
}
// 定义响应消息
message HelloResponse {
string result = 1;
}
// 定义服务 要想远程调用就必须定义服务接口
service HelloService {
// rpc 代表远程调用
// sum 代表远程调用的方法
// Request 代表远程调用要传入的参数
// Response 代表返回的结果
rpc hellRpcTest(HelloRequest) returns(HelloResponse) {}
}
输入mvn compile生成代码,如下图：
image-20210604140517140
打开HelloServiceGrpc，在代码中可以看到以下代码:
/**
* 创建一个异步的stub，支持服务的所有通信类型
* 简单理解就是这个stub可以支持所有的通信类型
*/
public static HelloServiceStub newStub(io.grpc.Channel channel) {
io.grpc.stub.AbstractStub.StubFactory<HelloServiceStub> factory =
new io.grpc.stub.AbstractStub.StubFactory<HelloServiceStub>() {
@java.lang.Override
public HelloServiceStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new HelloServiceStub(channel, callOptions);
}
};
return HelloServiceStub.newStub(factory, channel);
}
/**
* 创建一个阻塞的stu 支持 unary 通信和服务的流式输出
*/
public static HelloServiceBlockingStub newBlockingStub(
io.grpc.Channel channel) {
io.grpc.stub.AbstractStub.StubFactory<HelloServiceBlockingStub> factory =
new io.grpc.stub.AbstractStub.StubFactory<HelloServiceBlockingStub>() {
@java.lang.Override
public HelloServiceBlockingStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new HelloServiceBlockingStub(channel, callOptions);
}
};
return HelloServiceBlockingStub.newStub(factory, channel);
}
/**
* 创建一个新的 ListenableFuture 风格的存根，支持对服务的一元调用
*/
public static HelloServiceFutureStub newFutureStub(
io.grpc.Channel channel) {
io.grpc.stub.AbstractStub.StubFactory<HelloServiceFutureStub> factory =
new io.grpc.stub.AbstractStub.StubFactory<HelloServiceFutureStub>() {
@java.lang.Override
public HelloServiceFutureStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
return new HelloServiceFutureStub(channel, callOptions);
}
};
return HelloServiceFutureStub.newStub(factory, channel);
}
同时生成的还有一个基础服务类，如下
image-20210604141446523
1.1.2 server
创建maven模块(grpc-server),其pom.xml如下
<?xml version="1.0" encoding="UTF-8"?>
<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">
<parent>
<artifactId>grpc-parent</artifactId>
<groupId>com.tomato.wangzh</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>grpc-server</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>com.tomato.wangzh</groupId>
<artifactId>grpc-common</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>
创建GrpcService,内容如下：
package com.tomato.wangzh.grpc.server;
import com.tomato.wangzh.grpc.common.HelloRequest;
import com.tomato.wangzh.grpc.common.HelloResponse;
import com.tomato.wangzh.grpc.common.HelloServiceGrpc;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import java.util.UUID;
// Grpc 必须继承 HelloServiceImplBase
public class GrpcService extends HelloServiceGrpc.HelloServiceImplBase {
/**
* gprc远程调用 需要重写该方法，这个方法是在 proto文件里面定义的
* @param request 数据请求
* @param responseObserver 响应
*/
@Override
public void hellRpcTest(HelloRequest request, StreamObserver<HelloResponse> responseObserver) {
try {
// 获取请求参数
String param = request.getParam();
UUID uuid = UUID.randomUUID();
// 创建响应对象
HelloResponse response = HelloResponse.newBuilder().setResult(param + uuid.toString()).build();
// 设置响应对象
responseObserver.onNext(response);
// 告诉客户端处理完成
responseObserver.onCompleted();
} catch (Exception e) {
e.printStackTrace();
// 异常处理
responseObserver.onError(Status.DATA_LOSS.withDescription(e.getMessage()).asException());
}
}
}
封装一个Grpc的服务类GrpcServer用来进行测试，内容如下：
package com.tomato.wangzh.grpc.server;
import com.tomato.wangzh.grpc.common.HelloServiceGrpc;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
public class GrpcServer {
private static final Logger logger = Logger.getLogger(GrpcServer.class.getName());
/**
* grpc server port
*/
private Integer port;
/**
* grpc server
*/
private Server server;
/**
* grpc service
*/
private HelloServiceGrpc.HelloServiceImplBase service;
public GrpcServer(Integer port,HelloServiceGrpc.HelloServiceImplBase service) {
this(port,ServerBuilder.forPort(port),service);
}
public GrpcServer(Integer port, ServerBuilder builder,HelloServiceGrpc.HelloServiceImplBase service) {
logger.info("init grpc server");
this.port = port;
this.service = service;
server = builder.addService(service).build();
}
public void start() throws IOException {
server.start();
logger.info("start server");
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
logger.info("shut down grpc server because JVM is shuts down");
try {
GrpcServer.this.stop();
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.info("grpc server shut down");
}));
}
/**
* 关闭服务器，如果服务器关闭达到超时时间，则放弃
* @throws InterruptedException
*/
public void stop() throws InterruptedException {
if (server != null) {
// shut down grpc server
server.shutdown().awaitTermination(30, TimeUnit.SECONDS);
}
}
/**
* 阻塞服务器，直到服务器关闭
* @throws InterruptedException
*/
public void blockUntilShutdown() throws InterruptedException {
if (server != null) {
// blocking grpc server
server.awaitTermination();
}
}
public static void main(String[] args) {
try {
GrpcServer server = new GrpcServer(13452, new GrpcService());
server.start();
server.blockUntilShutdown();
} catch (IOException | InterruptedException e) {
e.printStackTrace();
// 启动服务器失败
logger.info("failed to start server");
}
}
}
1.1.3 client
创建maven模块grpc-client,其pom.xml内容如下：
<?xml version="1.0" encoding="UTF-8"?>
<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">
<parent>
<artifactId>grpc-parent</artifactId>
<groupId>com.tomato.wangzh</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>grpc-client</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>com.tomato.wangzh</groupId>
<artifactId>grpc-common</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty</artifactId>
</dependency>
</dependencies>
</project>
创建GrpcClient用来连接服务端，如下：
package com.tomato.wangzh.client;
import com.tomato.wangzh.grpc.common.HelloRequest;
import com.tomato.wangzh.grpc.common.HelloResponse;
import com.tomato.wangzh.grpc.common.HelloServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
public class GrpcClient {
private static final Logger logger = Logger.getLogger(GrpcClient.class.getName());
// 与服务端通信通道
private ManagedChannel channel;
// 阻塞stub 用于服务端通信
private HelloServiceGrpc.HelloServiceBlockingStub stub;
public GrpcClient(Integer port, String ip) {
this.channel =
ManagedChannelBuilder
.forAddress(ip, port)
.usePlaintext() // 使用文本传输 这个是最简的方式
.build(); // 构建通信channel
this.stub = HelloServiceGrpc.newBlockingStub(channel);
logger.info("init client");
}
/**
* 等待通道关闭直到达到超时时间放弃
* @throws InterruptedException
*/
public void shutdown() throws InterruptedException {
if (channel != null) {
channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
logger.info("shut down client channel");
}
}
public static void main(String[] args) throws InterruptedException {
GrpcClient client = null;
try {
// 创建客户端
client = new GrpcClient(13452, "localhost");
// 构建请求参数
HelloRequest request = HelloRequest.newBuilder().setParam("hello world").build();
// 远程调用方法
HelloResponse response = client.stub.hellRpcTest(request);
// 输出结果
System.out.println(response.getResult());
} finally {
if (client != null) {
// 关闭通道
client.shutdown();
}
}
}
}
1.1.4 测试
正常测试
先启动服务端如下：
image-20210604161709307
再启动客户端：
image-20210604161804655
可以看到客户端已经收到消息了
测试发送数据为 null
修改客户端测试代码，将请求参数设为null,如下：
image-20210604162104918
结果可以发现在客户端抛出了空指针异常
image-20210604162327424
测试服务端耗时操作
修改服务端代码，客户端代码修改正常
image-20210604162643823
重新启动客户端和服务端，可以发现客户端会进行等待，直到服务端返回信息
image-20210604162949834
这肯定是不行的，因此需要对耗时操作进行处理，修改请求等待时间，如下：
image-20210604163624109
启动后发现5s后没有响应，那么就直接报错，如下：
image-20210604164406183
注意：虽然客户端因为超时关闭了，但是这个请求还是会在服务端运行，这肯定是不允许的，一旦客户端取消了连接，那么该请求在服务端也应该取消。
在服务端增加去校代码，如下:
image-20210604173012699
再次启动客户端和服务端，如下：
image-20210604173641858
说明配置成功
1.2 Client Streaming
这种通信模型主要是客户端通过流式发送多次rpc请求给服务端，服务端对客户端进行一个简单的响应
image-20210606152020932
例如当客户端需要上传一个文件到服务端时就可以采用这种方式，客户端将文件拆分成以一个的小文件以流的方式顺序发送给服务端
服务端将这些文件接收到合并，这样比直接上传一个文件，从效率来说明显要更加快一点
因此我们就来实现是一个这样的案例，以此学习这种通信模型
1.2.1 common
在之前的grpc-common模块中继续增加protobuf文件(client_streaming.proto)，内容如下：
syntax = "proto3";
package client_streaming;
option java_multiple_files = true;
option java_package = "com.tomato.wangzh.grpc.common";
// 上传文件请求信息
message UploadFileRequest {
// 一条包含许多字段的消息，并且最多同时设置一个字段，您可以使用其中oneof功能来强制执行此行为并节省内存
oneof data {
FileInfo file_info = 1;
bytes file_data = 2;
}
}
// 创建文件信息
message UploadFileResponse {
string file_id = 1;
string size = 2;
}
// 文件信息
message FileInfo {
// 文件id
string file_id = 1;
// 文件类型
string file_type = 2;
}
service UploadFileService {
// stream 代表 客户端流
rpc uploadFile(stream UploadFileRequest) returns (UploadFileResponse) {}
}
输入mvn clean compile生成代码，如下：
image-20210607133805328
1.2.2 client
客户端以顺序流上传一张图片到服务端，服务端保存图片信息，具体流程如下：
image-20210616170005220
代码如下：
package com.tomato.wangzh.client.upload;
import com.google.protobuf.ByteString;
import com.tomato.wangzh.grpc.common.FileInfo;
import com.tomato.wangzh.grpc.common.UploadFileRequest;
import com.tomato.wangzh.grpc.common.UploadFileResponse;
import com.tomato.wangzh.grpc.common.UploadFileServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
import java.io.InputStream;
import java.text.MessageFormat;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
public class GrpcUploadClient {
private static final Logger logger = Logger.getLogger(GrpcUploadClient.class.getName());
private ManagedChannel channel;
// 注意： 与 helloword例子不同，这个用的是UploadFileServiceStub 而不是阻塞式的
private UploadFileServiceGrpc.UploadFileServiceStub stub;
public GrpcUploadClient(Integer port, String ip) {
channel = ManagedChannelBuilder.forAddress(ip, port).usePlaintext().build();
stub = UploadFileServiceGrpc.newStub(channel);
}
public static void main(String[] args) throws IOException {
GrpcUploadClient client = null;
CountDownLatch countDownLatch = new CountDownLatch(1);
// 读取文件数据
InputStream in = GrpcUploadClient.class.getResourceAsStream("/file/1.jpg");
StreamObserver<UploadFileRequest> requestObserver = null;
try {
// 创建客户端
client = new GrpcUploadClient(65534, "127.0.0.1");
// 构建requestObserver 用来传输数据
requestObserver =
client.stub.withDeadlineAfter(5,TimeUnit.SECONDS).uploadFile(new StreamObserver<UploadFileResponse>() {
@Override
public void onNext(UploadFileResponse value) {
// 获取服务端响应的结果
logger.log(Level.INFO, MessageFormat.format("响应的文件id: {0}", value.getFileId()));
logger.log(Level.INFO, MessageFormat.format("响应的文件大小: {0}", value.getSize()));
}
@Override
public void onError(Throwable t) {
logger.log(Level.SEVERE, MessageFormat.format("接受数据异常: {0}", t));
countDownLatch.countDown();
}
@Override
public void onCompleted() {
logger.log(Level.INFO, "上传完成");
countDownLatch.countDown();
}
});
// 构建文件信息
FileInfo fileInfo = FileInfo.newBuilder().setFileType("jpg").setFileId(UUID.randomUUID().toString()).build();
// 根据流程图，先发送文件信息
UploadFileRequest request = UploadFileRequest.newBuilder().setFileInfo(fileInfo).build();
// 发送文件信息
requestObserver.onNext(request);
// 构建数组
byte[] bytes = new byte[1024];
while(true) {
int count = in.read(bytes);
// 如果数据读完，那么就跳出循环
if (count <= 0) {
break;
}
// 如果传送结束或者发生异常，则结束整个方法
if (countDownLatch.getCount() == 0) {
return;
}
request = UploadFileRequest.newBuilder().setFileData(ByteString.copyFrom(bytes)).build();
requestObserver.onNext(request);
logger.info(MessageFormat.format("send data size : {0}", count) );
}
// 执行该方法代表传送完成
requestObserver.onCompleted();
// 如果两秒之内没有传送完成，则发生异常
if (countDownLatch.await(1, TimeUnit.MINUTES)) {
logger.info("Within one minute, the data was not transmitted");
}
} catch (Exception e) {
e.printStackTrace();
logger.log(Level.SEVERE, "failed to transfer data");
if (requestObserver != null) {
requestObserver.onError(e);
}
} finally {
if (client != null) {
client.shutdown();
logger.log(Level.INFO, "shuts down grpc client");
}
if (in != null) {
in.close();
}
}
}
public void shutdown() {
if (channel != null) {
channel.shutdown();
}
}
}
1.2.3 server
服务端接收到客户端的数据，但是对服务端来说并不知道发送过来的数据流是文件信息还是具体的文件数据，因此在获取是需要通过DataCase进行区分处理
image-20210617102825617
代码如下：
上传文件服务(UploadService)
package com.tomato.wangzh.grpc.server.upload;
import com.tomato.wangzh.grpc.common.UploadFileRequest;
import com.tomato.wangzh.grpc.common.UploadFileResponse;
import com.tomato.wangzh.grpc.common.UploadFileServiceGrpc;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import io.netty.util.internal.StringUtil;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.text.MessageFormat;
import java.util.Objects;
import java.util.Random;
import java.util.logging.Level;
import java.util.logging.Logger;
public class GrpcUploadService extends UploadFileServiceGrpc.UploadFileServiceImplBase {
private static final Logger logger = Logger.getLogger(GrpcUploadService.class.getName());
@Override
public StreamObserver<UploadFileRequest> uploadFile(StreamObserver<UploadFileResponse> responseObserver) throws FileNotFoundException {
return new StreamObserver<UploadFileRequest>() {
/**
* 获取客户端发送过来的数据
* @param value
*/
long size = 0;
// 构建文件输出流
Integer id = new Random().nextInt(100);
FileOutputStream out = new FileOutputStream(id + ".jpg");
@Override
public void onNext(UploadFileRequest value) {
try {
// 1.获取发送过来的流类型
UploadFileRequest.DataCase dataCase = value.getDataCase();
// 如果是文件信息则对文件信息处理
if (Objects.equals(dataCase, UploadFileRequest.DataCase.FILE_INFO)) {
String fileId = value.getFileInfo().getFileId();
String fileType = value.getFileInfo().getFileType();
logger.log(Level.INFO, MessageFormat.format("The id of the file sent by the client: {0}",fileId));
logger.log(Level.INFO, MessageFormat.format("The file type sent by the client: {0}",fileType));
} else if (Objects.equals(dataCase, UploadFileRequest.DataCase.FILE_DATA)){
size += value.getFileData().size();
out.write(value.getFileData().toByteArray());
}
} catch (IOException e) {
e.printStackTrace();
responseObserver.onError(Status.INTERNAL.withDescription("server write error").asRuntimeException());
}
}
@Override
public void onError(Throwable t) {
logger.log(Level.SEVERE,MessageFormat.format("client send data error: {0}", t));
}
@Override
public void onCompleted() {
try {
// 传输完成，则响应客户端一个信息
out.flush();
out.close();
UploadFileResponse response = UploadFileResponse.newBuilder().setFileId(id.toString()).setSize(size+"").build();
responseObserver.onNext(response);
responseObserver.onCompleted();
} catch (IOException e) {
e.printStackTrace();
responseObserver.onError(Status.DATA_LOSS.withDescription(e.getMessage()).asRuntimeException());
}
}
};
}
}
注意：因为在方法上抛出了异常，因此在父类方法中也需要抛出异常
image-20210617113519979
服务器(GrpcUploadServer)代码如下：
package com.tomato.wangzh.grpc.server.upload;
import com.tomato.wangzh.grpc.common.UploadFileServiceGrpc;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
public class GrpcServer {
public Server server;
private static final Logger logger = Logger.getLogger(GrpcServer.class.getName());
public GrpcServer(Integer port, UploadFileServiceGrpc.UploadFileServiceImplBase service) {
this(ServerBuilder.forPort(port), service);
}
public GrpcServer(ServerBuilder builder, UploadFileServiceGrpc.UploadFileServiceImplBase service) {
server = builder.addService(service).build();
}
public void start() throws IOException {
if (server != null) {
server.start();
}
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
GrpcServer.this.stop();
}));
}
public void stop() {
try {
if (server != null) {
server.shutdown().awaitTermination(5, TimeUnit.SECONDS);
}
} catch (InterruptedException e) {
e.printStackTrace();
logger.log(Level.SEVERE,"failed to start grpc server");
}
}
public void await() {
try {
if (server != null) {
server.awaitTermination();
}
} catch (InterruptedException e) {
e.printStackTrace();
logger.log(Level.SEVERE,"failed to start grpc server");
}
}
public static void main(String[] args) throws IOException {
GrpcServer server = new GrpcServer(65534, new GrpcUploadService());
server.start();
logger.info("started grpc server successfully");
server.await();
}
}
启动后，结果如下：
image-20210617113953153
可以看到客户端中得到了响应，服务端响应如下：
image-20210617114045537
同时也可以看到文件保存下来：
image-20210617114120711
自此 cleint-stream模型完成
1.2.4 测试
先后启动服务端和客户端，就会看到数据，客户端不断的发送数据过去，最后服务端只响应了一个结果
image-20210704203106384
1.3 Server Streaming
这种通信模型与Client Streaming很相似，客户端发送一个请求过去，服务端不断以流的形式不断返回数据给客户端，如下：
image-20210617141354843
例如，当客户端发送一个数字给服务端，服务端不停的返回该数字的倍数给客户端，直到返回的数是大于100为止。这样就可以用这种通信模型。
这种案例就是服务端不停的返回数据给客户端，用Server Streaming这种通信模型刚刚好
1.3.1. common
在之前的grpc-common模块中继续增加protobuf文件(server_streaming.proto)，内容如下：
syntax = "proto3";
package server_streaing;
option java_multiple_files = true;
option java_package = "com.tomato.wangzh.grpc.common";
// 定义请求的消息
message Request {
uint32 number = 1;
}
// 定义响应的消息
message Response {
uint32 result = 1;
}
// 定义请求
service ServerStreamingService {
rpc getResult(Request) returns (stream Response);
}
1.3.2 server
服务端主要是接收客户端发送过来的数字，并且以流的形式返回多个数据给客户端，如下图所示：
image-20210705173210552
在grpc-server中定义NumberService,用来处理给客户端返回消息，具体代码如下：
package com.tomato.wangzh.grpc.server.number;
import com.tomato.wangzh.grpc.common.Request;
import com.tomato.wangzh.grpc.common.Response;
import com.tomato.wangzh.grpc.common.ServerStreamingServiceGrpc;
import io.grpc.Context;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
public class NumberService extends ServerStreamingServiceGrpc.ServerStreamingServiceImplBase {
@Override
public void getResult(Request request, StreamObserver<Response> responseObserver) {
int number = request.getNumber();
if (Context.current().isCancelled()) {
responseObserver.onError(Status.CANCELLED.withDescription("request is cancelled").asRuntimeException());
return;
}
for(int i = 1; i <= 100; i++) {
if (i % number == 0) {
Response response = Response.newBuilder().setResult(i).build();
responseObserver.onNext(response);
}
}
responseObserver.onCompleted();
}
}
定义GRPCServer服务器，如下：
package com.tomato.wangzh.grpc.server.number;
import com.tomato.wangzh.grpc.common.ServerStreamingServiceGrpc;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
public class GRPCSever {
public static Logger logger = Logger.getLogger(GRPCSever.class.getName());
public static Server server;
public GRPCSever(Integer port, ServerStreamingServiceGrpc.ServerStreamingServiceImplBase service) {
this(ServerBuilder.forPort(port),service);
}
public GRPCSever(ServerBuilder serverBuilder,ServerStreamingServiceGrpc.ServerStreamingServiceImplBase service) {
server = serverBuilder.addService(service).build();
}
public void start() throws IOException {
server.start();
logger.info("started server");
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
GRPCSever.this.stop();
}));
}
public void stop() {
if (server != null) {
server.shutdown();
}
}
public void await() throws InterruptedException {
if (server != null) {
server.awaitTermination();
}
}
public static void main(String[] args) throws InterruptedException, IOException {
GRPCSever grpcSever = new GRPCSever(65533, new NumberService());
grpcSever.start();
grpcSever.await();
}
}
1.3.3 client
客户端发送一个数字给服务端，服务端接受到数字以后，返回该数字的倍数，直到数大于100，如下图所示:
image-20210705175333391
在grpc-client新建GRPCClient,代码如下：
package com.tomato.wangzh.client.number;
import com.tomato.wangzh.grpc.common.Request;
import com.tomato.wangzh.grpc.common.Response;
import com.tomato.wangzh.grpc.common.ServerStreamingServiceGrpc;
import io.grpc.Channel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import java.text.MessageFormat;
import java.util.logging.Level;
import java.util.logging.Logger;
public class GRPCClient {
public static final Logger logger = Logger.getLogger(GRPCClient.class.getName());
private Channel channel = null;
/**
* 只要不是简单通信模型，都是采用这种类型的stub
*/
private ServerStreamingServiceGrpc.ServerStreamingServiceStub stub;
public GRPCClient(String ip,Integer port) {
// usePlaintext 代表使用普通文本传输，不采用加密方式
channel = ManagedChannelBuilder.forAddress(ip,port).usePlaintext().build();
stub = ServerStreamingServiceGrpc.newStub(channel);
}
public static void main(String[] args) {
GRPCClient grpcClient = new GRPCClient("127.0.0.1", 65533);
grpcClient.stub.getResult(Request.newBuilder().setNumber(5).build(), new StreamObserver<Response>() {
@Override
public void onNext(Response response) {
logger.info(MessageFormat.format("服务端发送过来的数字:{0}" , response.getResult()));
}
@Override
public void onError(Throwable throwable) {
logger.log(Level.WARNING,MessageFormat.format("数据发送错误:{0}",throwable.getMessage()));
}
@Override
public void onCompleted() {
logger.info("数据传送完成");
}
});
}
}
1.3.4 测试
先后启动服务端和客户端，可以看到客户端接受到了多次结果，如下：
[图片上传失败...(image-1d5d9-1625559692582)]
1.4 Bidirectional Streaming
这种通信模型相当于Client Streaming 与 Server Streaming的结合体，因此这种模型也叫双向流模型
即客户端以流的形式发送数据给服务端，而服务端也是以流的形式响应数据数据给客户端
如下图所示:
image-20210706095614162
实际开发过程中，经常会有以下场景:
客户端发送多个商品id向服务端查询商品，服务端接收客户端的id，并且去数据库查询商品
这种场景下，如果采用之前的的三种模型，在一次请求中做到基本上是不太可能，而如果采用Bidirectional Streaming则是可以做到
因此接下来采用该模型来实现上述案例
1.4.1 common
还是在之前的项目的common模块，增加bidirectional_streaming.proto文件，内容如下：
syntax = "proto3";
package bidirectional_streaming;
// 代表生成java文件在哪个包底下
option java_package = "com.tomato.wangzh.grpc.common";
// 代表生成多个文件
option java_multiple_files = true;
// 定义请求消息
message ProductRequest {
string id = 1;
}
// 定义响应消息
message ProductResponse {
Product product = 1;
}
// 定义响应消息内容
message Product {
string id = 1;
string name = 2;
double price = 3;
}
// 定义远程调用服务
service ProductService {
rpc getProductById(stream ProductRequest) returns(stream ProductResponse) {}
}
跟之前一样，通过输入mvn clean compile生成代码
1.4.2 server
服务端用来接受客户端发送过来的多个id，并且进行处理响应，如图：
image-20210706105530327
在之前的grpc-server中，新增ProductService.java用来处理获取商品服务，内容如下：
package com.tomato.wangzh.grpc.server.product;
import com.tomato.wangzh.grpc.common.Product;
import com.tomato.wangzh.grpc.common.ProductRequest;
import com.tomato.wangzh.grpc.common.ProductResponse;
import com.tomato.wangzh.grpc.common.ProductServiceGrpc;
import io.grpc.Context;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import java.text.MessageFormat;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;
public class ProductService extends ProductServiceGrpc.ProductServiceImplBase {
/**
* 模拟数据库
*/
public static final Map<String, Product> map = new ConcurrentHashMap<String,Product>(){{
put("12343",Product.newBuilder().setId("12343").setName("衣服").setPrice(53.2).build());
put("45678",Product.newBuilder().setId("45678").setName("裤子").setPrice(65.5).build());
put("78797",Product.newBuilder().setId("78797").setName("鞋子").setPrice(33.3).build());
put("91067",Product.newBuilder().setId("91067").setName("帽子").setPrice(898).build());
}};
public static final Logger logger = Logger.getLogger(ProductService.class.getName());
@Override
public StreamObserver<ProductRequest> getProductById(StreamObserver<ProductResponse> responseObserver) {
return new StreamObserver<ProductRequest>() {
@Override
public void onNext(ProductRequest productRequest) {
// 接收客户端消息
logger.info(MessageFormat.format("接受客户端的数据:{0}",productRequest.getId()));
// 处理客户端取消连接问题
if (Context.ROOT.isCancelled()) {
responseObserver.onError(Status.CANCELLED.withDescription("客户端取消了链接").asRuntimeException());
responseObserver.onCompleted();
return;
}
// 根据id查询数据
Product product = map.get(productRequest.getId());
// 响应客户端数据，接受一个数据响应一个数据
ProductResponse response = ProductResponse.newBuilder().setProduct(product).build();
responseObserver.onNext(response);
}
@Override
public void onError(Throwable throwable) {
logger.log(Level.WARNING,"接受数据发生了异常");
responseObserver.onError(Status.CANCELLED.withDescription(MessageFormat.format("接受数据发生了异常:{0}",throwable.getMessage())).asRuntimeException());
responseObserver.onCompleted();
}
@Override
public void onCompleted() {
// 数据接受完成，即响应完成
responseObserver.onCompleted();
}
};
}
}
接下来构建BidirectionalServer.java用来接受客户端发送过来的数据，代码如下：
package com.tomato.wangzh.grpc.server.product;
import com.tomato.wangzh.grpc.common.ProductServiceGrpc;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
public class BidirectionalServer {
private Server server;
public static final Logger logger = Logger.getLogger(BidirectionalServer.class.getName());
public BidirectionalServer(Integer port, ProductServiceGrpc.ProductServiceImplBase service) {
this(ServerBuilder.forPort(port),service);
}
public BidirectionalServer(ServerBuilder builder, ProductServiceGrpc.ProductServiceImplBase service) {
server = builder.addService(service).build();
}
/**
* 启动方法
*/
public void start() {
if (server != null) {
try {
server.start();
Runtime.getRuntime().addShutdownHook(new Thread(BidirectionalServer.this::stop));
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 阻塞方法
*/
public void stop() {
if (server != null) {
try {
server.shutdown().awaitTermination(3, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/**
* 阻塞方法
*/
public void await() {
if (server != null) {
try {
server.awaitTermination();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
BidirectionalServer server = new BidirectionalServer(65511,new ProductService());
server.start();
server.await();
}
}
1.4.3 client
客户端主要是向服务端发送多个商品id，并且接受服务端返回来的商品数据，如下图所示：
image-20210706111917063
在grpc-client中新增ProductClient.java,用来向服务端发送数据以及接收付服务端的数据，内容如下:
package com.tomato.wangzh.client.product;
import com.tomato.wangzh.grpc.common.Product;
import com.tomato.wangzh.grpc.common.ProductRequest;
import com.tomato.wangzh.grpc.common.ProductResponse;
import com.tomato.wangzh.grpc.common.ProductServiceGrpc;
import io.grpc.Channel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import static java.text.MessageFormat.format;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.logging.Logger;
public class ProductClient {
/**
* 连接通道
*/
private Channel channel;
/**
* 通信的客户端
*/
private ProductServiceGrpc.ProductServiceStub stub;
private static final Logger logger = Logger.getLogger(ProductClient.class.getName());
/**
* 存放多个商品id
*/
private List<String> ids = new CopyOnWriteArrayList<String>(){{
add("12343");
add("91067");
}};
public ProductClient(String ip, Integer port) {
channel = ManagedChannelBuilder.forAddress(ip, port).usePlaintext().build();
stub = ProductServiceGrpc.newStub(channel);
}
public static void main(String[] args) throws InterruptedException {
ProductClient client = new ProductClient("127.0.0.1", 65511);
CountDownLatch countDownLatch = new CountDownLatch(1);
// 得到发送服务端消息的流
StreamObserver<ProductRequest> requestStreamObserver = client.stub.getProductById(new StreamObserver<ProductResponse>() {
@Override
public void onNext(ProductResponse productResponse) {
// 接受服务端发送过来的消息
Product product = productResponse.getProduct();
logger.info(format("product信息:\n\tid:{0},name:{1},price:{1}", product.getId(), product.getName(), product.getPrice()));
}
@Override
public void onError(Throwable throwable) {
logger.warning(format("接受服务端消息发生错误:{0}", throwable.getMessage()));
countDownLatch.countDown();
}
@Override
public void onCompleted() {
countDownLatch.countDown();
}
});
client.ids.forEach(t -> {
// 不停的向服务端发送数据
ProductRequest request = ProductRequest.newBuilder().setId(t).build();
requestStreamObserver.onNext(request);
});
// 循环结束代表发送完成
requestStreamObserver.onCompleted();
countDownLatch.await();
}
}
1.4.4 测试
先后启动服务端和客户端，结果如下：
image-20210706113819308
自此四种通信模型就全部了解完毕，实际开发中使用什么模型需要取决于具体的通信模型
网友评论