04-gRPC java 版

一般来讲，实现一个gRPC服务端和客户端，主要分为这几步：

安装 protobuf 依赖
编写 proto 文件（IDL）
编译 proto 文件（生成stub文件）
编写 server 端，实现我们的接口
编写 client 端，测试我们的接口

1、第一个 gPRC 的开发

项目结构：

xxxx-api 模块
1. 定义 protobuf IDL 语言
2. 并且通过命令创建对应的代码
xxxx-server 模块
1. 实现 api 模块中定义的服务接口
2. 发布 gRPC 服务 (创建服务端程序)
xxxx-clien 模块
1. 创建服务端 stub(代理)
2. 基于代理（stub) RPC 调用。

目录结构

├── grpc-api
│   ├── pom.xml
│   ├── src
├── grpc-client
│   ├── pom.xml
│   ├── src
├── grpc-server
│   ├── pom.xml
│   ├── src
└── pom.xml

api 模块

编写 `.proto` 文件，书写 protobuf 的 IDL。

syntax = "proto3";

option java_multiple_files = false;                     // 后续 protobuf 生成的java代码 一个源文件还是多个源文件  xx.java
option java_package = "com.lepeng.grpcserver.grpc";     // 指定 protobuf 生成的类 放置在哪个包中
option java_outer_classname = "HelloProto";             // 指定 protobuf 生成的外部类的名字（管理内部类【内部类才是真正开发使用】）

service HelloService{
    rpc hello(HelloRequest) returns (HelloResponse){}
    rpc hello1(HelloRequest1) returns (HelloResponse1){}
}

把 IDL 转化为编程语言，有两种方式：

使用 protoc 命令把 proto 文件中的 IDL 转换成编程语言
protoc --java_out=/xxx/xxx /xxx/xxx/xx.proto
使用 maven 插件，进行 protobuf IDL 文件的编译，并把他放置 IDEA 具体位置。

<project>
    <dependencies>
        <dependency>
            <groupId>io.grpc</groupId>
            <artifactId>grpc-netty-shaded</artifactId>
            <version>1.52.1</version>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>io.grpc</groupId>
            <artifactId>grpc-protobuf</artifactId>
            <version>1.52.1</version>
        </dependency>
        <dependency>
            <groupId>io.grpc</groupId>
            <artifactId>grpc-stub</artifactId>
            <version>1.52.1</version>
        </dependency>
        <dependency> <!-- necessary for Java 9+ -->
            <groupId>org.apache.tomcat</groupId>
            <artifactId>annotations-api</artifactId>
            <version>6.0.53</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>
    
    <build>
        <extensions>
            <extension>
                <groupId>kr.motd.maven</groupId>
                <artifactId>os-maven-plugin</artifactId>
                <version>1.7.1</version>
            </extension>
        </extensions>
        <plugins>
            <plugin>
                <groupId>org.xolstice.maven.plugins</groupId>
                <artifactId>protobuf-maven-plugin</artifactId>
                <version>0.6.1</version>
                <configuration>
                    <protocArtifact>com.google.protobuf:protoc:3.21.7:exe:${os.detected.classifier}</protocArtifact>
                    <pluginId>grpc-java</pluginId>
                    <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.52.1:exe:${os.detected.classifier}</pluginArtifact>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>  <!-- compile 用来编译消息对象 -->
                            <goal>compile-custom</goal>  <!-- compile-custom 则依赖消息对象,生成接口服务 -->
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

在 idea 中分别点击 compile 和 compile-custom。compile 和 compile-custom 两个指令都需要执行。其中

compile 用来编译消息对象，生成消息体类文件
compile-custom 则依赖消息对象，生成接口服务，生成XXXGrpc类文件。

HelloProto 的默认生成目录为 target/generated-sources/protobuf/grpc，放的是生成的 message 对应的 java 对象
HelloServiceGrpc 的默认生成目录为 target/generated-sources/protobuf/grpc-java，放的是生成的 Service 对应的类

将转化的代码放到对应目录，即 com.lepeng.grpcserver 子目录 grpc 下

xxxx-server 服务端模块的开发

实现业务接口添加具体的功能（MyBatis+MySQL)

开发步骤

构建监听地址 SocketAddress
1. SPI 加载 NettyServerProvider
2. 根据指定端口创建监听地址
将 Service 注册到缓存
Server 构建
服务端启动

原理参考

package com.lepeng.grpcserver.grpc.server;

import com.lepeng.grpcserver.grpc.HelloServiceGrpc;
import com.lepeng.grpcserver.grpc.HelloProto.HelloResponse;
import com.lepeng.grpcserver.grpc.HelloProto.HelloRequest;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;

import java.io.IOException;
import java.util.logging.Logger;

public class HelloServer {
    private static final Logger logger = Logger.getLogger(HelloServer.class.getName());
    private int port = 50051;
    private Server server;

    private void start() throws IOException {
        // 使用 ServerBuilder 来构建和启动服务，通过使用 forPort 方法来指定监听的地址和端口
        // 创建一个实现方法的服务 HelloServiceImpl 的实例，并通过 addService 方法将该实例纳入
        // 调用 build() start() 方法构建和启动 rpcserver
        server = ServerBuilder.forPort(port)
                .addService(new HelloServiceImpl())
                .build()
                .start();
        logger.info("Server started, listening on " + port);

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public 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");
                HelloServer.this.stop();
                System.err.println("*** server shut down");
            }
        });
    }

    private void stop() {
        if (server != null) {
            server.shutdown();
        }
    }

    /**
     * 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 HelloServer server = new HelloServer();
        server.start();
        server.blockUntilShutdown();
    }


    // 我们的服务 HelloServiceImpl 继承了生成抽象类 HelloServiceGrpc.HelloServiceImplBase，实现了服务的所有方法
    public class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase {
        /*
        1. 接受client提交的参数  request.getParameter()
        2. 业务处理 service+dao 调用对应的业务功能。
        3. 提供返回值
        */
        @Override
        public void hello(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) {
            // 1.接受client的请求参数
            String name = request.getName();
            // 2.业务处理
            System.out.println("name parameter "+name);
            
            // 3.封装响应
            // 3.1 创建相应对象的构造者
            HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
            // 3.2 填充数据
            builder.setResult("hello method invoke ok");
            // 3.3 封装响应
            HelloProto.HelloResponse helloResponse = builder.build();
            
            // 4.使用响应监视器的 onNext 方法返回 helloResponse
            responseObserver.onNext(helloResponse);
            // 5.使用 onCompleted 方法指定本次调用已经完成
            responseObserver.onCompleted();
        }
    }
}

Netty 创建服务端

public class GprcServer1 {
    public static void main(String[] args) throws IOException, InterruptedException {
        //1. 绑定端口 
        ServerBuilder serverBuilder = ServerBuilder.forPort(9000);
        //2. 发布服务
        serverBuilder.addService(new HelloServiceImpl());
        //serverBuilder.addService(new UserServiceImpl());
        //3. 创建服务对象
        Server server = serverBuilder.build();
        
        server.start();
        server.awaitTermination();;
    }
}

xxx-client 模块

client 通过代理对象完成远端对象的调用

package com.lepeng.grpcclient.grpc.client;

import com.lepeng.grpcclient.grpc.HelloServiceGrpc;
import com.lepeng.grpcclient.grpc.HelloProto.HelloResponse;
import com.lepeng.grpcclient.grpc.HelloProto.HelloRequest;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.StatusRuntimeException;

import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

public class GprcClient1 {
    public static void main(String[] args) {
        // 1 首先，我们需要为 stub 创建一个 grpc 的 channel，指定我们连接服务端的地址和端口
        // 使用 ManagedChannelBuilder 方法来创建 channel
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build();
        //2 获得代理对象 stub
        try {
            // 使用我们从 proto 文件生成的 HelloServiceGrpc 类提供的 newBlockingStub 方法指定 channel 创建 stubs
            HelloServiceGrpc.HelloServiceBlockingStub helloService = HelloServiceGrpc.newBlockingStub(managedChannel);
            
            //3. 完成RPC调用
            //3.1 准备参数
            HelloProto.HelloRequest.Builder builder = HelloProto.HelloRequest.newBuilder();
            builder.setName("this is name");
            HelloProto.HelloRequest helloRequest = builder.build();
            //3.1 进行功能rpc调用，获取相应的内容
            HelloProto.HelloResponse helloResponse = helloService.hello(helloRequest);
            String result = helloResponse.getResult();
            System.out.println("result = " + result);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }finally {
            managedChannel.shutdown();
        }
    }
}

注意事项

服务端 处理返回值时
responseObserver.onNext(helloResponse1);  // 通过这个方法 把响应的消息 回传 client
responseObserver.onCompleted();           // 通知 client 整个服务结束。底层返回标记 
                                          // client 就会监听标记 【grpc做的】
                                        
requestObserver.onNext(helloRequest1);
requestObserver.onCompleted();

执行

启动服务端，后台打印输出：Server started, listening on 50051

执行客户端，后台打印输出：

1 2	`Will try to greet hans ... Greeting: Hello hans`

gRpc 的四种通信方式

四种通信方式

简单 RPC，一元 RPC (Unary RPC)
服务端流式 RPC (Server Streaming RPC)
客户端流式 RPC (Client Streaming RPC)
双向流 RPC (Bi-directional Stream RPC)

简单RPC(一元RPC)

上面的 RPC 程序，实际上就是一元 RPC

特点：当 client 发起调用后，提交数据，并且等待服务端响应。开发过程中，主要采用就是一元 RPC 的这种通信方式

protobuf

service HelloService{
    rpc hello(HelloRequest) returns (HelloResponse){}
    rpc hello1(HelloRequest1) returns (HelloResponse1){}
}

服务端流式 RPC

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

特点

使用场景

client  --------> Server
        股票标号
        <-------
         某一个时刻的 股票的行情

protobuf

service HelloService{
    rpc hello(HelloRequest) returns (stream HelloResponse){}
    rpc hello1(HelloRequest1) returns (HelloResponse1){}
}

代码

// 服务端
public void c2ss(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) {
    //1 接受client的请求参数
    String name = request.getName();
    //2 做业务处理
    System.out.println("name = " + name);
    //3 根据业务处理的结果，提供响应
    for (int i = 0; i < 9; i++) {
        HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
        builder.setResult("处理的结果 " + i);
        HelloProto.HelloResponse helloResponse = builder.build();

        responseObserver.onNext(helloResponse);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
    responseObserver.onCompleted();
}


// 客户端
public class GprcClient3 {
    public static void main(String[] args) {
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build();
        try {
            HelloServiceGrpc.HelloServiceBlockingStub helloService = HelloServiceGrpc.newBlockingStub(managedChannel);

            HelloProto.HelloRequest.Builder builder = HelloProto.HelloRequest.newBuilder();
            builder.setName("sunshuai");
            HelloProto.HelloRequest helloRequest = builder.build();
            Iterator<HelloProto.HelloResponse> helloResponseIterator = helloService.c2ss(helloRequest);
            while (helloResponseIterator.hasNext()) {
                HelloProto.HelloResponse helloResponse = helloResponseIterator.next();
                System.out.println("helloResponse.getResult() = " + helloResponse.getResult());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        finally {
            managedChannel.shutdown();
        }
    }
}

//监听 异步方式 处理服务端流式RPC的开发
//1. api
//2. 服务端 
//3. 客户端 
public class GrpcClient4 {
    public static void main(String[] args) {
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build();

        try {

            HelloServiceGrpc.HelloServiceStub helloService = HelloServiceGrpc.newStub(managedChannel);

            HelloProto.HelloRequest.Builder builder = HelloProto.HelloRequest.newBuilder();
            builder.setName("xiaohei");
            HelloProto.HelloRequest helloRequest = builder.build();

            helloService.c2ss(helloRequest, new StreamObserver<HelloProto.HelloResponse>() {
                @Override
                public void onNext(HelloProto.HelloResponse value) {
                    //服务端 响应了 一个消息后，需要立即处理的话。把代码写在这个方法中。
                    System.out.println("服务端每一次响应的信息 " + value.getResult());
                }

                @Override
                public void onError(Throwable t) {

                }

                @Override
                public void onCompleted() {
                    //需要把服务端 响应的所有数据 拿到后，在进行业务处理。
                    System.out.println("服务端响应结束 后续可以根据需要 在这里统一处理服务端响应的所有内容");
                }
            });

            managedChannel.awaitTermination(12, TimeUnit.SECONDS);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            managedChannel.shutdown();
        }
    }
}

客户端流式 RPC

客户端发送多个请求对象，服务端只返回一个结果

应用场景：IOT(物联网【传感器】) 向服务端发送数据

protobuf

1	`rpc cs2s(stream HelloRequest) returns (HelloResponse){}`

开发

// 1. api
rpc cs2s(stream HelloRequest) returns (HelloResponse){}


// 2. 服务端开发
public StreamObserver<HelloProto.HelloRequest> cs2s(StreamObserver<HelloProto.HelloResponse> responseObserver) {
    return new StreamObserver<HelloProto.HelloRequest>() {
        @Override
        public void onNext(HelloProto.HelloRequest value) {
            System.out.println("接受到了client发送一条消息 " + value.getName());
        }

        @Override
        public void onError(Throwable t) {

        }

        @Override
        public void onCompleted() {
            System.out.println("client的所有消息 都发送到了 服务端 ....");

            //提供响应：响应的目的：当接受了全部client提交的信息，并处理后，提供相应
            HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
            builder.setResult("this is result");
            HelloProto.HelloResponse helloResponse = builder.build();

            responseObserver.onNext(helloResponse);
            responseObserver.onCompleted();
        }
    };
}


// 3. 客户端开发
public class GrpcClient5 {
    public static void main(String[] args) {
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build();
        try {
            HelloServiceGrpc.HelloServiceStub helloService = HelloServiceGrpc.newStub(managedChannel);

            StreamObserver<HelloProto.HelloRequest> helloRequestStreamObserver = helloService.cs2s(new StreamObserver<HelloProto.HelloResponse>() {
                @Override
                public void onNext(HelloProto.HelloResponse value) {
                    // 监控响应
                    System.out.println("服务端 响应 数据内容为 " + value.getResult());

                }

                @Override
                public void onError(Throwable t) {

                }

                @Override
                public void onCompleted() {
                    System.out.println("服务端响应结束 ... ");
                }
            });

            //客户端 发送数据 到服务端  多条数据 ，不定时...
            for (int i = 0; i < 10; i++) {
                HelloProto.HelloRequest.Builder builder = HelloProto.HelloRequest.newBuilder();
                builder.setName("sunshuai " + i);
                HelloProto.HelloRequest helloRequest = builder.build();

                helloRequestStreamObserver.onNext(helloRequest);
                Thread.sleep(1000);
            }

            helloRequestStreamObserver.onCompleted();

            managedChannel.awaitTermination(12, TimeUnit.SECONDS);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            managedChannel.shutdown();
        }
    }
}

双向流式 RPC

客户端可以发送多个请求消息，服务端响应多个响应消息

应用场景：聊天室

编码

// 1. api
rpc cs2ss(stream HelloRequest) returns (stream HelloResponse){}


//2. 服务端
public StreamObserver<HelloProto.HelloRequest> cs2ss(StreamObserver<HelloProto.HelloResponse> responseObserver) {
    return new StreamObserver<HelloProto.HelloRequest>() {
         @Override
         public void onNext(HelloProto.HelloRequest value) {
             System.out.println("接受到client 提交的消息 "+value.getName());
             responseObserver.onNext(HelloProto.HelloResponse.newBuilder().setResult("response "+value.getName()+" result ").build());
         }

         @Override
         public void onError(Throwable t) {

         }

         @Override
         public void onCompleted() {
             System.out.println("接受到了所有的请求消息 ... ");
             responseObserver.onCompleted();
         }
    };
}


//3. 客户端
public class GrpcClient6 {
    public static void main(String[] args) {
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build();
        try {
            HelloServiceGrpc.HelloServiceStub helloService = HelloServiceGrpc.newStub(managedChannel);

            StreamObserver<HelloProto.HelloRequest> helloRequestStreamObserver = helloService.cs2ss(new StreamObserver<HelloProto.HelloResponse>() {
                @Override
                public void onNext(HelloProto.HelloResponse value) {
                    System.out.println("响应的结果 "+value.getResult());
                }

                @Override
                public void onError(Throwable t) {

                }

                @Override
                public void onCompleted() {
                    System.out.println("响应全部结束...");
                }
            });


            for (int i = 0; i < 10; i++) {
                helloRequestStreamObserver.onNext(HelloProto.HelloRequest.newBuilder().setName("sunshuai " + i).build());
            }
            helloRequestStreamObserver.onCompleted();

            managedChannel.awaitTermination(12, TimeUnit.SECONDS);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            managedChannel.shutdown();
        }
    }
}

gPRC 代理方式

BlockingStub：阻塞通信方式
Stub：异步通过监听处理的
FutureStub：同步异步 NettyFuture
1. FutureStub 只能应用一元RPC

public class GrpcClient7 {
    public static void main(String[] args) {
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build();
        try {
            TestServiceGrpc.TestServiceFutureStub testServiceFutureStub = TestServiceGrpc.newFutureStub(managedChannel);
            ListenableFuture<TestProto.TestResponse> responseListenableFuture = testServiceFutureStub.testSuns(TestProto.TestRequest.newBuilder().setName("xiaojren").build());


            /* 同步操作
            TestProto.TestResponse testResponse = responseListenableFuture.get();
            System.out.println(testResponse.getResult());*/

            /*  responseListenableFuture.addListener(() -> {
                System.out.println("异步的rpc响应 回来了....");
            }, Executors.newCachedThreadPool());*/

            Futures.addCallback(responseListenableFuture, new FutureCallback<TestProto.TestResponse>() {
                @Override
                public void onSuccess(TestProto.TestResponse result) {
                    System.out.println("result.getResult() = " + result.getResult());
                }

                @Override
                public void onFailure(Throwable t) {

                }
            }, Executors.newCachedThreadPool());


            System.out.println("后续的操作....");

            managedChannel.awaitTermination(12, TimeUnit.SECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            managedChannel.shutdown();
        }
    }
}

gPRC 与 SpringBoot 整合

gRPC 和 SpringBoot 整合的思想

grpc-server
grpc-client

SpringBoot 与 GRPC 整合的过程中对于服务端做了什么封装

搭建 SpringBoot 的开发环境

引入与 Grpc 相关的内容

<dependency>
    <groupId>com.suns</groupId>
    <artifactId>rpc-grpc-api</artifactId>
    <version>1.0-SNAPSHOT</version>
</dependency>

<dependency>
  <groupId>net.devh</groupId>
  <artifactId>grpc-server-spring-boot-starter</artifactId>
  <version>2.14.0.RELEASE</version>
</dependency>

开发服务

// 重复 多次 
@GrpcService
public class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase {
    @Override
    public void hello(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) {
        String name = request.getName();
        System.out.println("name is " + name);

        responseObserver.onNext(HelloProto.HelloResponse.newBuilder().setResult("this is result").build());
        responseObserver.onCompleted();
    }
}

// application.yml
# 核心配置的 就是gRPC服务的端口号
spring:
  application:
    name: boot-server

  main:
    web-application-type: none

grpc:
  server:
    port: 9000

客户端

环境搭建

<dependency>
    <groupId>net.devh</groupId>
    <artifactId>grpc-client-spring-boot-starter</artifactId>
    <version>2.14.0.RELEASE</version>
</dependency>

编码

1. yml
grpc:
client:
  grpc-server:
    address: 'static://127.0.0.1:9000'
    negotiation-type: plaintext
    
2. 注入stub

@GrpcClient("grpc-server")
private HelloServiceGrpc.HelloServiceBlockingStub stub;

gRPC

#gRPC

04-gRPC java 版

https://flepeng.github.io/046-gRPC-04-gRPC-java-版/

作者

Lepeng

发布于

2024年4月1日

许可协议

04-gRPC java 插件上一篇

00-kubeflow pipeline dsl 下一篇

04-gRPC java 版

1、第一个 gPRC 的开发

api 模块

编写 .proto 文件，书写 protobuf 的 IDL。

把 IDL 转化为编程语言，有两种方式：

xxxx-server 服务端模块的开发

xxx-client 模块

注意事项

执行

gRpc 的四种通信方式

简单RPC(一元RPC)

服务端流式 RPC

客户端流式 RPC

双向流式 RPC

gPRC 代理方式

gPRC 与 SpringBoot 整合

gRPC 和 SpringBoot 整合的思想

SpringBoot 与 GRPC 整合的过程中 对于服务端做了什么封装

客户端

编写 `.proto` 文件，书写 protobuf 的 IDL。

SpringBoot 与 GRPC 整合的过程中对于服务端做了什么封装