做设计拍摄的网站平台,重庆网站建设推广,北京建设工程招标信息网站,网站建设新方式简单省时一、 简介
gRPC的定义#xff1a;
一个高性能、通用的开源RPC框架主要面向移动应用开发#xff1a; gRPC提供了一种简单的方法来精确地定义服务和为iOS、Android和后台支持服务自动生成可靠性很强的客户端功能库。基于HTTP/2协议标准而设计#xff0c;基于ProtoBuf(Protoc…一、 简介
gRPC的定义
一个高性能、通用的开源RPC框架主要面向移动应用开发 gRPC提供了一种简单的方法来精确地定义服务和为iOS、Android和后台支持服务自动生成可靠性很强的客户端功能库。基于HTTP/2协议标准而设计基于ProtoBuf(Protocol Buffers)序列化协议开发支持众多开发语言二、 简单rpc调用
主要流程
创建maven项目添加grpc依赖,protobuf依赖和插件通过.proto文件定义服务通过protocol buffer compiler插件生成客户端和服务端通过grpc API生成客户端和服务端代码
1. 创建maven项目
添加pom依赖, 包含依赖包和生成基础类的插件。
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdvip.sunjin/groupIdartifactIdGrpcServer/artifactIdversion1.0-SNAPSHOT/versionpropertiesproject.build.sourceEncodingUTF-8/project.build.sourceEncodingproject.reporting.outputEncodingUTF-8/project.reporting.outputEncodingjava.version1.8/java.versiongrpc.version1.36.1/grpc.versionprotobuf.version3.15.6/protobuf.version/propertiesdependencies!-- protobuf --dependencygroupIdcom.google.protobuf/groupIdartifactIdprotobuf-java/artifactIdversion${protobuf.version}/version/dependency!-- GRPC --dependencygroupIdio.grpc/groupIdartifactIdgrpc-netty/artifactIdversion${grpc.version}/version/dependencydependencygroupIdio.grpc/groupIdartifactIdgrpc-protobuf/artifactIdversion${grpc.version}/version/dependencydependencygroupIdio.grpc/groupIdartifactIdgrpc-stub/artifactIdversion${grpc.version}/version/dependency/dependenciesbuildextensionsextensiongroupIdkr.motd.maven/groupIdartifactIdos-maven-plugin/artifactIdversion1.6.2/version/extension/extensionspluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/pluginplugingroupIdorg.xolstice.maven.plugins/groupIdartifactIdprotobuf-maven-plugin/artifactIdversion0.6.1/versionconfigurationprotocArtifactcom.google.protobuf:protoc:3.12.0:exe:${os.detected.classifier}/protocArtifactpluginIdgrpc-java/pluginIdpluginArtifactio.grpc:protoc-gen-grpc-java:1.36.0:exe:${os.detected.classifier}/pluginArtifact/configurationexecutionsexecutiongoalsgoalcompile/goalgoalcompile-custom/goal/goals/execution/executions/plugin/plugins/build/project
2. 定义RPC服务数据结构 proto文件
创建一个文件夹src/main/proto/
创建一个helloworld.proto文件
syntax proto3;option java_multiple_files true;
option java_package vip.sunjin.examples.helloworld;
option java_outer_classname HelloWorldProto;package helloworld;// The greeting service definition.
service Greeter {// Sends a greetingrpc SayHello (HelloRequest) returns (HelloReply) {}
}// The request message containing the users name.
message HelloRequest {string name 1;
}// The response message containing the greetings
message HelloReply {string message 1;
}3. 生成基础类
使用maven编译项目 生成基础类
生成后的类文件如下
targetclassesvipsunjinexampleshelloworldGreeterGrpc.javaHelloReply.javaHelloReplyOrBuilder.javaHelloRequest.javaHelloRequestOrBuilder.javaHelloWorldClient.javaHelloWorldProto.javaHelloWorldServer.java
GreeterGrpc封装基本的GRPC功能后续的客户端和服务端都从这个类引申出来。
4. 创建服务端
服务端只需要指定一个端口号然后暴露一个服务。
/*** Server that manages startup/shutdown of a {code Greeter} server.*/
public class HelloWorldServer {private static final Logger logger Logger.getLogger(HelloWorldServer.class.getName());private Server server;private void start() throws IOException {/* The port on which the server should run */int port 50051;server ServerBuilder.forPort(port).addService(new GreeterImpl()).build().start();logger.info(Server started, listening on port);Runtime.getRuntime().addShutdownHook(new Thread() {Overridepublic void run() {// Use stderr here since the logger may have been reset by its JVM shutdown hook.System.err.println(*** shutting down gRPC server since JVM is shutting down);try {HelloWorldServer.this.stop();} catch (InterruptedException e) {e.printStackTrace(System.err);}System.err.println(*** server shut down);}});}private void stop() throws InterruptedException {if (server ! null) {server.shutdown().awaitTermination(30, TimeUnit.SECONDS);}}/*** Await termination on the main thread since the grpc library uses daemon threads.*/private void blockUntilShutdown() throws InterruptedException {if (server ! null) {server.awaitTermination();}}/*** Main launches the server from the command line.*/public static void main(String[] args) throws IOException, InterruptedException {final HelloWorldServer server new HelloWorldServer();server.start();server.blockUntilShutdown();}static class GreeterImpl extends GreeterGrpc.GreeterImplBase {Overridepublic void sayHello(HelloRequest req, StreamObserverHelloReply responseObserver) {HelloReply reply HelloReply.newBuilder().setMessage(Hello req.getName()).build();responseObserver.onNext(reply);responseObserver.onCompleted();}}
} 5. 创建客户端
客户端需要指定调用服务的地址和端口号并且通过调用桩代码调用服务端的服务。
客户端和服务端是直连的。
public class HelloWorldClient {private static final Logger logger Logger.getLogger(HelloWorldClient.class.getName());private final GreeterGrpc.GreeterBlockingStub blockingStub;/** Construct client for accessing HelloWorld server using the existing channel. */public HelloWorldClient(Channel channel) {// channel here is a Channel, not a ManagedChannel, so it is not this codes responsibility to// shut it down.// Passing Channels to code makes code easier to test and makes it easier to reuse Channels.blockingStub GreeterGrpc.newBlockingStub(channel);}/** Say hello to server. */public void greet(String name) {logger.info(Will try to greet name ...);HelloRequest request HelloRequest.newBuilder().setName(name).build();HelloReply response;try {response blockingStub.sayHello(request);} catch (StatusRuntimeException e) {logger.log(Level.WARNING, RPC failed: {0}, e.getStatus());return;}logger.info(Greeting Reply: response.getMessage());}/*** Greet server. If provided, the first element of {code args} is the name to use in the* greeting. The second argument is the target server.*/public static void main(String[] args) throws Exception {String user neil;// Access a service running on the local machine on port 50051String target localhost:50051;// Create a communication channel to the server, known as a Channel. Channels are thread-safe// and reusable. It is common to create channels at the beginning of your application and reuse// them until the application shuts down.ManagedChannel channel ManagedChannelBuilder.forTarget(target)// Channels are secure by default (via SSL/TLS). For the example we disable TLS to avoid// needing certificates..usePlaintext().build();try {HelloWorldClient client new HelloWorldClient(channel);client.greet(user);} finally {// ManagedChannels use resources like threads and TCP connections. To prevent leaking these// resources the channel should be shut down when it will no longer be used. If it may be used// again leave it running.channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);}}
}
6. 测试
先启动服务端代码 HelloWorldServer
然后执行客户端代码 HelloWorldClient
执行结果如下 2月 18, 2023 9:44:17 上午 vip.sunjin.grpcclient.HelloWorldClient greet 信息: Will try to greet neil ... 2月 18, 2023 9:44:18 上午 vip.sunjin.grpcclient.HelloWorldClient greet 信息: Greeting Reply: Im Grpc Server , Hello neil 三、 grpc服务端流
一般业务场景下我们都是使用grpc的simple-rpc模式也就是每次客户端发起请求服务端会返回一个响应结果的模式。
但是grpc除了这种一来一往的请求模式外还有流式模式。
服务端流模式是说客户端发起一次请求后服务端在接受到请求后可以以流的方式使用同一连接不断的向客户端写回响应结果客户端则可以源源不断的接受到服务端写回的数据。
下面我们通过简单例子来说明如何使用服务端端流。
1. 定义RPC服务数据结构 proto文件
MetricsService.proto
syntax proto3;option java_multiple_files true;
option java_package vip.sunjin.examples.helloworld;
option java_outer_classname MetricsServiceProto;message Metric {int64 metric 2;
}message Average {double val 1;
}service MetricsService {rpc collectServerStream (Metric) returns (stream Average);
}
然后使用maven编译项目 生成基础类
2.创建服务端代码
服务实现类
public class MetricsServiceImpl extends MetricsServiceGrpc.MetricsServiceImplBase {private static final Logger logger Logger.getLogger(MetricsServiceImpl.class.getName());/*** 服务端流* param request* param responseObserver*/Overridepublic void collectServerStream(Metric request, StreamObserverAverage responseObserver) {logger.info(received request : request.getMetric());for(int i 0; i 10; i){responseObserver.onNext(Average.newBuilder().setVal(new Random(1000).nextDouble()).build());logger.info(send to client);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}responseObserver.onCompleted();}}
服务端Server启动类
public class MetricsServer {private static final Logger logger Logger.getLogger(MetricsServer.class.getName());public static void main(String[] args) throws IOException, InterruptedException {int port 50051;
// //启动服务MetricsServiceImpl metricsService new MetricsServiceImpl();Server server ServerBuilder.forPort(port).addService(metricsService).build();server.start();logger.info(Server started, listening on port);server.awaitTermination();}
}
3.创建客户端代码
通过异步Stub 调用服务
public class MetricsClient {private static final Logger logger Logger.getLogger(MetricsClient.class.getName());public static void main(String[] args) throws InterruptedException {int port 50051;
// //获取客户端桩对象ManagedChannel channel ManagedChannelBuilder.forTarget(localhost: port).usePlaintext().build();MetricsServiceGrpc.MetricsServiceStub stub MetricsServiceGrpc.newStub(channel);//发起rpc请求设置StreamObserver用于监听服务器返回结果stub.collectServerStream(Metric.newBuilder().setMetric(1L).build(), new StreamObserverAverage() {Overridepublic void onNext(Average value) {System.out.println(Thread.currentThread().getName() Average: value.getVal());}Overridepublic void onError(Throwable t) {System.out.println(error: t.getLocalizedMessage());}Overridepublic void onCompleted() {System.out.println(onCompleted:);}});channel.shutdown().awaitTermination(50, TimeUnit.SECONDS);}
}
代码最后要有等待并且关闭通道的操作。
4.测试
先启动服务端再启动客户端后可以看到StreamObserver的onNext方法会源源不断的接受到服务端返回的数据。
5服务端流使用场景
客户端请求一次但是需要服务端源源不断的返回大量数据时候比如大批量数据查询的场景。比如客户端订阅服务端的一个服务数据服务端发现有新数据时源源不断的吧数据推送给客户端。四、 grpc客户端流
客户端流模式是说客户端发起请求与服务端建立链接后可以使用同一连接不断的向服务端传送数据等客户端把全部数据都传送完毕后服务端才返回一个请求结果。
1. 定义RPC服务数据结构 proto文件
这里修改service的定义其他不变。 MetricsService.proto
service MetricsService {rpc collectClientStream (stream Metric) returns (Average);
}
2.创建服务端代码
如上rpc方法的入参类型前添加stream标识 是客户端流然后服务端实现代码如下
public class MetricsServiceImpl extends MetricsServiceGrpc.MetricsServiceImplBase {private static final Logger logger Logger.getLogger(MetricsServiceImpl.class.getName());/*** 客户端流** param responseObserver* return*/Overridepublic StreamObserverMetric collectClientStream(StreamObserverAverage responseObserver) {return new StreamObserverMetric() {private long sum 0;private long count 0;Overridepublic void onNext(Metric value) {logger.info(value: value);sum value.getMetric();count;}Overridepublic void onError(Throwable t) {logger.info(severError: t.getLocalizedMessage());responseObserver.onError(t);}Overridepublic void onCompleted() {responseObserver.onNext(Average.newBuilder().setVal(sum / count).build());logger.info(serverComplete: );responseObserver.onCompleted();}};}
}
如上代码服务端使用流式对象的onNext方法不断接受客户端发来的数据然后等客户端发送结束后使用onCompleted方法把响应结果写回客户端。
服务端启动类MetricsServer不需要修改 3.创建客户端代码
客户端调用服务需要使用异步的Stub. public class MetricsClient2 {private static final Logger logger Logger.getLogger(MetricsServer.class.getName());public static void main(String[] args) throws InterruptedException {int port 50051;//1.创建客户端桩ManagedChannel channel ManagedChannelBuilder.forAddress(localhost, port).usePlaintext().build();MetricsServiceGrpc.MetricsServiceStub stub MetricsServiceGrpc.newStub(channel);//2.发起请求并设置结果回调监听StreamObserverMetric collect stub.collectClientStream(new StreamObserverAverage() {Overridepublic void onNext(Average value) {logger.info(Thread.currentThread().getName() Average: value.getVal());}Overridepublic void onError(Throwable t) {logger.info(error: t.getLocalizedMessage());}Overridepublic void onCompleted() {logger.info(onCompleted:);}});//3.使用同一个链接不断向服务端传送数据Stream.of(1L, 2L, 3L, 4L,5L).map(l - Metric.newBuilder().setMetric(l).build()).forEach(metric - {collect.onNext(metric);logger.info(send to server: metric.getMetric());});Thread.sleep(3000);collect.onCompleted();channel.shutdown().awaitTermination(50, TimeUnit.SECONDS);}
}
4.测试
先启动服务端再启动客户端后可以看到代码3会把数据12345通过同一个链接发送到服务端 然后等服务端接收完毕数据后会计算接受到的数据的平均值然后把平均值写回客户端。 然后代码2设置的监听器的onNext方法就会被回调然后打印出服务端返回的平均值3。 5客户端流使用场景
比如数据批量计算场景如果只用simple rpc的话服务端就要一次性收到大量数据并且在收到全部数据之后才能对数据进行计算处理。如果用客户端流 rpc的话服务端可以在收到一些记录之后就开始处理也更有实时性。五、grpc双向流
双向流意味着客户端向服务端发起请求后客户端可以源源不断向服务端写入数据的同时服务端可以源源不断向客户端写入数据。
1. 定义RPC服务数据结构 proto文件
这里修改service的定义其他不变。 重新生成基础代码。 MetricsService.proto
service MetricsService {rpc collectTwoWayStream (stream Metric) returns (stream Average);
}
如上rpc方法的入参类型前添加stream标识, 返回参数前也添加stream标识 就是双向流然后服务端实现代码如下
双向流的代码和客户端流基本一样只是双向流可以同时支持双向的持续写入。
2.创建服务端代码
将服务实现类进行修改用来测试双向流。
public void onNext(Metric value) {logger.info(value: value);sum value.getMetric();count;responseObserver.onNext(Average.newBuilder().setVal(sum * 1.0/ count).build());
}
如上代码服务端使用流式对象的onNext方法不断接受客户端发来的数据 然后 不断的调用参数中的流对象把响应结果持续的写回客户端。 实现了双向流式调用。
3.创建客户端代码
客户端代码主要是把onCompleted之前的线程等待时间加长以便等待服务端持续的返回。
Thread.sleep(10000);
collect.onCompleted();
4.测试
先启动服务端再启动客户端后可以看到代码3会把数据12345通过同一个链接发送到服务端 然后等服务端接收数据后会实时的计算接受到的数据的平均值然后把平均值写回客户端。 然后代码2设置的监听器的onNext方法就会被回调然后打印出服务端返回的平均值。
