一 需求

根据 id 查 name

二 编写 proto 文件

syntax = "proto3"; package grpc.proto; option java_package = "com.grpc.proto"; option java_outer_classname = "StudentData"; option java_multiple_files = true ; // 定义接口 service StudentService { // 请求一个 Requset 对象，响应一个 Response 对象 rpc queryStudentNameById(MyRequestId) returns(MyResponseName) {} // 请求一个 Requset 对象，响应一个 Stream 对象 rpc queryStudentsByCourseName(MyRequestCourseName) returns(stream MyResponseStudentsStream) {} // 请求一个 Stream 对象，响应一个 Response 对象 rpc queryStudentsByCourseName2(stream MyRequestCourseName) returns(MyResponseStudents) {} // 请求一个 Stream，响应一个 Stream 对象，本例测试这个接口 rpc queryStudentNameById2(stream MyRequestId) returns(stream MyResponseName) {} } message MyRequestId { int32 id = 1 ; } message MyResponseName { string name = 1 ; } message MyStudent { int32 id = 1 ; string name = 2; string courseName = 3 ; } message MyResponseStudents { // 服务端的响应结果是集合类型，因此需要加上 repeated repeated MyStudent students = 1 ; } // 数据结构,定义请求的 Request 对象 message MyRequestCourseName { string courseName = 1 ; } // 数据结构,定义响应的 Stream message MyResponseStudentsStream { int32 id = 1 ; string name = 2; string courseName = 3 ; }

三 编写接口实现类

package grpc; import grpc.proto.*; import io.grpc.stub.StreamObserver; public class StudentServiceImpl extends StudentServiceGrpc.StudentServiceImplBase { @Override public StreamObserver queryStudentNameById2(StreamObserver responseObserver) { MyStreamObserver2 observer = new MyStreamObserver2(); observer.setResponseObserver(responseObserver); return observer; } class MyStreamObserver2 implements StreamObserver { private StreamObserver responseObserver; private MyResponseName responseStudentName; public void setResponseObserver(StreamObserver responseObserver) { this.responseObserver = responseObserver; } @Override public void onNext(MyRequestId value) { System.out.println("接收到的请求参数是：" + value.getId()); // 假设查到的结果是“zs” this.responseStudentName = MyResponseName.newBuilder().setName("zs").build(); } @Override public void onError(Throwable t) { t.printStackTrace(); } @Override public void onCompleted() { responseObserver.onNext(responseStudentName); responseObserver.onCompleted(); } } }

四 编写服务端代码

package grpc; import io.grpc.Server; import io.grpc.ServerBuilder; import java.io.IOException; public class MyGRPCServer { private Server server; // 启动服务 private void start() throws IOException { int port = 8888; server = ServerBuilder.forPort(port) .addService(new StudentServiceImpl()) .build() .start(); Runtime.getRuntime().addShutdownHook(new Thread(() ->{ System.err.println(Thread.currentThread().getName() + "，关闭JVM"); // 当 JVM 关闭时，也同时关闭 MyGRPCServer服 务 MyGRPCServer.this.stop(); } )); } // 关闭服务 private void stop() { if (server != null) { server.shutdown(); } } private void blockUntilShutdown() throws InterruptedException { if (server != null) { // 等待服务结束 server.awaitTermination(); } } public static void main(String[] args) throws IOException, InterruptedException { final MyGRPCServer server = new MyGRPCServer(); server.start(); server.blockUntilShutdown(); } }

五 编写客户端代码

package grpc; import grpc.proto.MyRequestId; import grpc.proto.MyResponseName; import grpc.proto.StudentServiceGrpc; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import io.grpc.stub.StreamObserver; public class MyGRPCClient { public static void main(String[] args) throws Exception { // 创建一个客户端 ManagedChannel client = ManagedChannelBuilder.forAddress("127.0.0.1", 8888) .usePlaintext().build(); // 在 grpc 中，如果是以 Stream 方式发出请求，则此请求是异步的。因此，不能再使用阻塞式 stub 对象。 StudentServiceGrpc.StudentServiceStub stub = StudentServiceGrpc .newStub(client); // 请求一个 Stream，响应一个 Stream StreamObserver requestIdObserver = stub.queryStudentNameById2(new StreamObserver() { @Override public void onNext(MyResponseName value) { System.out.println("接收到的响应：" + value.getName()); } @Override public void onError(Throwable t) { t.printStackTrace(); } @Override public void onCompleted() { System.out.println("查询结束"); } }); requestIdObserver.onNext(MyRequestId.newBuilder().setId(1).build()); requestIdObserver.onCompleted(); Thread.sleep(3000); client.shutdown(); } }

六 测试

1 启动服务端

2 启动客户端

3 服务端打印

接收到的请求参数是：1

4 客户端打印

接收到的响应：zs

查询结束