[网络协议]gRPC Java入门示例

1. gRPC介绍

随着云原生时代的到来：

• K8s作为事实上的标准Pass底座
• Istio作为未来的微服务框架（Service Mesh）
• Go作为流行的云原生语言（K8s、Istio）
• Istio对Http、gRPC的原生支持
• 以及最近的Dubbo3 Triple协议全面兼容gRPC

是时候注意到gRPC这个关键词了，
gRPC起源于Google的微服务RPC框架Stubby，后在2015年3月由Google开源为当前的gRPC，

• 目前支持11种程序开发语言（跨语言）
• 使用proto作为IDL接口定义语言（即通过proto定义接口及数据，官方推荐proto3）
• 基于HTTP2实现（原生支持Http2 双向流通信）
• 支持插件式的auth, tracing, load balancing, health checking

gRPC通过Protobuf（官方推荐proto3）定义RPC服务的相关接口：

• 服务
• 方法
• 参数类型
• 返回结果类型

使用proto3定义gRPC服务示例（GreeterService.proto）：

// RPC服务定义
service HelloService {
  //RPC方法定义
  rpc SayHello (HelloRequest) returns (HelloResponse);
}

// 请求参数定义
message HelloRequest {
  string name = 1;
}

//响应结果定义
message HelloReply {
  string message = 1;
}

然后使用protoc（需安装gRPC插件）或者 后文提到的maven插件 根据预先定义的*.proto文件生成：

• RPC Client端代码(Stub) - 用于客户端调用
• RPC Server端代码 - 服务端需实现接口逻辑，并且启动gRPC server
• 参数、返回结果的相关Protobuf对象
• 支持Synchronous vs. asynchronous两种模式

gRPC支持4种方法类型

• Unary RPCs - 一元的RPC，单独的request和response

rpc SayHello(HelloRequest) returns (HelloResponse);

• Server streaming RPCs - 服务端流RPC，仅发送一个request，然后由Server端返回response流（源源不断的response）直到再无response，由gRPC保证response消息顺序。

rpc LotsOfReplies(HelloRequest) returns (stream HelloResponse);

• Client streaming RPCs - 客户端流RPC，Client端发送request消息流直到再无请request，然后Server端仅返回一个response消息，由gRPC保证request消息顺序

rpc LotsOfGreetings(stream HelloRequest) returns (HelloResponse);

• Bidirectional streaming RPCs - 双向流RPC，双向的读写(request, response)流，且request和response流可各自独立保持消息顺序，例如Server端在接受全部request后才统一发送response，或者接到一条request后就发送response，又或者其他任意组合。

rpc BidiHello(stream HelloRequest) returns (stream HelloResponse);

gRPC目前支持的语言见下表：

proto3支持的field类型见下表：

2. 核心概念

RPC(Remote Procedure Call)
远程过程调用，客户端就像调用本地方法一样调用远程服务，例如通过接口定义进行调用。

IDL(Interface Definition Language)
接口定义语言，定义服务：

• 服务
• 方法
• 参数类型
• 返回结果类型

Protobuf(protocol buffers)
一种结构化数据的系列化方法，
在gRPC中可用于服务接口及方法的参数和返回值，
亦可用于网络编程中的数据通信。

Stub
特定语言的Client端，用于调用Server端服务，与Server端具有相同方法定义

3. gRPC Java入门示例

结合官方提供的HelloWorld入门示例，并丰富gRPC方法类型示例，
按照如下步骤创建入门示例grpc-demo，源码可参见：https://gitee.com/luoex/grpc-demo.git

3.1 maven依赖

gRPC运行依赖：

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <grpc.version>1.44.0</grpc.version>
    <protobuf.version>3.19.2</protobuf.version>
    <protoc.version>3.19.2</protoc.version>
    <gson.version>2.8.9</gson.version>
</properties>

<!-- gRPC公共依赖管理 -->
<dependencyManagement>
    <dependencies>
        <!-- gRPC bom -->
        <dependency>
            <groupId>io.grpc</groupId>
            <artifactId>grpc-bom</artifactId>
            <version>${grpc.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <!-- protobuf依赖 -->
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java-util</artifactId>
            <version>${protobuf.version}</version>
        </dependency>
        <!-- prevent downgrade via protobuf-java-util -->
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>${gson.version}</version>
        </dependency>
    </dependencies>
</dependencyManagement>


<dependencies>
    <!-- gRPC依赖 -->
    <dependency>
        <groupId>io.grpc</groupId>
        <artifactId>grpc-netty-shaded</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>io.grpc</groupId>
        <artifactId>grpc-protobuf</artifactId>
    </dependency>
    <dependency>
        <groupId>io.grpc</groupId>
        <artifactId>grpc-stub</artifactId>
    </dependency>

    <!-- protobuf依赖 -->
    <dependency>
        <groupId>com.google.protobuf</groupId>
        <artifactId>protobuf-java-util</artifactId>
    </dependency>
    <!-- prevent downgrade via protobuf-java-util -->
    <dependency>
        <groupId>com.google.code.gson</groupId>
        <artifactId>gson</artifactId>
    </dependency>
</dependencies>

gRPC生成代码插件：

<build>
    <extensions>
        <!-- 兼容eclipse和netbeans中protobuf代码生成插件-->
        <extension>
            <groupId>kr.motd.maven</groupId>
            <artifactId>os-maven-plugin</artifactId>
            <version>1.6.2</version>
        </extension>
    </extensions>
    <plugins>
        <!-- grpc代码生成插件 -->
        <plugin>
            <groupId>org.xolstice.maven.plugins</groupId>
            <artifactId>protobuf-maven-plugin</artifactId>
            <version>0.6.1</version>
            <configuration>
                <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
                <pluginId>grpc-java</pluginId>
                <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
            </configuration>
            <executions>
                <execution>
                    <goals>
                        <goal>compile</goal>
                        <goal>compile-custom</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-enforcer-plugin</artifactId>
            <version>1.4.1</version>
            <executions>
                <execution>
                    <id>enforce</id>
                    <goals>
                        <goal>enforce</goal>
                    </goals>
                    <configuration>
                        <rules>
                            <requireUpperBoundDeps/>
                        </rules>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

使用如上protobuf-maven-plugin插件定义，即可：

• 在执行mvn compile时，
• 自动扫描代码目中src/main/proto下的proto文件，
• 然后自动生成gRPC相关代码到target/generated-sources/protobuf目录下。

3.2 定义proto

通过proto3定义gRPC服务，覆盖之前提到的几种方法类型：

具体proto文件src/main/proto/hello.proto定义如下：

/*
* HelloWorld入门示例
*/

//使用proto3语法
syntax = "proto3";

//proto包名
package hello;
//生成多个Java文件
option java_multiple_files = true;
//指定Java包名
option java_package = "com.luo.demo.grpc.hello";
//指定Java输出类名
option java_outer_classname = "HelloProto";



//gRPC服务定义
service Hello {
  //gRPC服务方法定义 - Unary
  rpc sayHello (HelloRequest) returns (HelloReply) {}

  //gRPC服务方法定义 - Server Streaming - 服务端流
  rpc sayHelloServerStream (HelloRequest) returns (stream HelloReply) {}

  //gRPC服务方法定义 - Client Streaming - 客户端流
  rpc sayHelloClientStream (stream HelloRequest) returns (HelloReply) {}

  //gRPC服务方法定义 - BiDirection Streaming - 双向流
  rpc sayHelloBiStream (stream HelloRequest) returns (stream HelloReply) {}
}

//请求参数定义
message HelloRequest {
  string name = 1;
}

//响应结果定义
message HelloReply {
  string message = 1;
}

3.3 生成代码

执行mvn compile后，会自动根据src/main/proto/hello.proto生成相关代码如下图：

将绿框中的代码（即grpc-java和java目录下）拷贝到程序中代码对应的包目录，如下图：

其中HelloGrpc即为gRPC服务的代码定义，其中包括：

• HelloGrpc.HelloImplBase - 服务端继承实现该类，对应具体的服务逻辑
• HelloGrpc.newBlockingSub, HelloGrpc.newStub - 用于客户端生成stub，即调用端

其他HelloRequest、HelloReply等即为具体的参数、结果的protobuf相关代码（提供Builder模式用于快速构建对象）。

3.4 gRPC Server端编码

首先Server端需要实现具体的服务逻辑，即继承实现HelloGrpc.HelloImplBase类，
然后启动gRPC Server，并注册服务端实现类。

对应不同的gRPC方法类型，具体HelloGrpcImpl实现代码如下：

import com.luo.demo.grpc.hello.HelloGrpc;
import com.luo.demo.grpc.hello.HelloReply;
import com.luo.demo.grpc.hello.HelloRequest;
import io.grpc.stub.StreamObserver;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * Hello gRPC服务 - 实现类
 *
 * @author luohq
 * @date 2022-02-06 13:46
 */
public class HelloGrpcImpl extends HelloGrpc.HelloImplBase {

    @Override
    public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
        System.out.printf("[sayHello]recv: %s", request);
        //构造返回结果
        HelloReply helloReply = HelloReply.newBuilder()
                .setMessage("Hello " + request.getName())
                .build();
        System.out.printf("[sayHello]resp: %s", helloReply);
        //输出响应
        responseObserver.onNext(helloReply);
        //结束响应
        System.out.println("[sayHello]resp completed!");
        responseObserver.onCompleted();
    }

    @Override
    public void sayHelloServerStream(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
        System.out.printf("[sayHelloServerStream]recv: %s", request);
        //服务端输出响应流（多次onNext输出响应结果）
        IntStream.range(0, 5).forEach(index -> {
            //构造返回结果
            HelloReply helloReply = HelloReply.newBuilder()
                    .setMessage(String.format("Hello_%d %s", index, request.getName()))
                    .build();
            System.out.printf("[sayHelloServerStream]resp: %s", helloReply);
            //输出响应
            responseObserver.onNext(helloReply);
        });

        //结束响应
        System.out.println("[sayHelloServerStream]resp completed!");
        responseObserver.onCompleted();
    }

    @Override
    public StreamObserver<HelloRequest> sayHelloClientStream(StreamObserver<HelloReply> responseObserver) {
        //实现StreamObserver接受客户端流
        return new StreamObserver<HelloRequest>() {
            //name列表
            private List<String> nameList = new ArrayList<>();

            @Override
            public void onNext(HelloRequest request) {
                //接受请求
                nameList.add(request.getName());
                System.out.printf("[sayHelloClientStream]recv_%d: %s\n", nameList.size(), request.getName());
            }

            @Override
            public void onError(Throwable t) {
                //处理错误
                System.err.println("[sayHelloClientStream]recv error!");
                t.printStackTrace();
            }

            @Override
            public void onCompleted() {
                //构造返回结果
                String nameListStr = nameList.stream().collect(Collectors.joining(","));
                HelloReply helloReply = HelloReply.newBuilder().setMessage(String.format("Hello %s", nameListStr)).build();
                System.out.printf("[sayHelloClientStream]resp: %s", helloReply);
                //输出响应
                responseObserver.onNext(helloReply);
                System.out.println("[sayHelloClientStream]resp completed!");
                //结束响应
                responseObserver.onCompleted();
            }
        };
    }

    @Override
    public StreamObserver<HelloRequest> sayHelloBiStream(StreamObserver<HelloReply> responseObserver) {
        //实现StreamObserver接受客户端流
        return new StreamObserver<HelloRequest>() {
            @Override
            public void onNext(HelloRequest request) {
                //接受请求
                System.out.printf("[sayHelloBiStream]recv: %s\n", request.getName());
                //构造返回结果
                HelloReply helloReply = HelloReply.newBuilder()
                        .setMessage("Hello " + request.getName())
                        .build();
                System.out.printf("[sayHelloBiStream]resp: %s", helloReply);
                //输出响应
                responseObserver.onNext(helloReply);
            }

            @Override
            public void onError(Throwable t) {
                //处理错误
                System.err.println("[sayHelloBiStream]recv error!");
                t.printStackTrace();
            }

            @Override
            public void onCompleted() {
                System.out.println("[sayHelloBiStream]resp completed!");
                //结束响应
                responseObserver.onCompleted();
            }
        };
    }
}

Server端核心启动代码如下：

//启动gRPC Server
Integer port = 50051;
Server server = ServerBuilder.forPort(port)
         //注册服务端实现类
        .addService(new HelloGrpcImpl())
        .build()
        .start();
System.out.println("gRPC Server started, listening on " + port);

3.5 gRPC Client端编码

客户端即构建连接服务端的channel，然后构建客户端调用stub，

核心代码如下：

//服务端地址
String target = "localhost:50051";
//构建channel
ManagedChannel managedChannel = ManagedChannelBuilder.forTarget(target)
                .usePlaintext()
                .build();
//阻塞Hello客户端（仅支持Unary、Server Stream）
HelloGrpc.HelloBlockingStub helloBlockingStub = HelloGrpc.newBlockingStub(managedChannel);
//非阻塞Hello客户端（全部仅支持Unary、Server Stream、Client Stream、BiDirection Stream）
HelloGrpc.HelloStub helloStub = HelloGrpc.newStub(managedChannel);

客户端的详细构建及调用逻辑实现代码HelloClient内容如下：

import com.luo.demo.grpc.hello.HelloGrpc;
import com.luo.demo.grpc.hello.HelloReply;
import com.luo.demo.grpc.hello.HelloRequest;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.util.Iterator;

/**
 * Hello gRPC服务 - 客户端
 *
 * @author luohq
 * @date 2022-02-06 13:45
 */
public class HelloClient {

    public static void main(String[] args) throws Exception {
        String target = "localhost:50051";

        //初始化client stub
        HelloClient helloClient = new HelloClient();
        helloClient.init(target);

        //调用服务测试
        helloClient.callSayHello();
        helloClient.callSayHelloServerStream();
        helloClient.callSayHelloClientStream();
        helloClient.callSayHelloBiStream();


        //阻塞主线程，等待gRPC客户端异步调用完成
        Thread.sleep(10000);
    }

    /**
     * 阻塞Hello客户端（仅支持Unary、Server Stream）
     */
    private HelloGrpc.HelloBlockingStub helloBlockingStub;
    /**
     * 非阻塞Hello客户端（全部仅支持Unary、Server Stream、Client Stream、BiDirection Stream）
     */
    private HelloGrpc.HelloStub helloStub;

    /**
     * 客户端初始化
     *
     * @param target 服务端连接目标
     */
    public void init(String target) {
        ManagedChannel managedChannel = ManagedChannelBuilder.forTarget(target)
                .usePlaintext()
                .build();
        this.helloBlockingStub = HelloGrpc.newBlockingStub(managedChannel);
        this.helloStub = HelloGrpc.newStub(managedChannel);
    }

    /**
     * 调用sayHello
     */
    public void callSayHello() {
        //构建请求参数
        HelloRequest helloRequest = HelloRequest.newBuilder()
                .setName("luo")
                .build();

        //阻塞API
        HelloReply helloReply = this.helloBlockingStub.sayHello(helloRequest);
        System.out.printf("[callSayHello]blocking resp: %s", helloReply);

        //非阻塞API
        this.helloStub.sayHello(helloRequest, new StreamObserver<HelloReply>() {
            @Override
            public void onNext(HelloReply value) {
                System.out.printf("[callSayHello]resp: %s", value);
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("[callSayHello]error");
            }

            @Override
            public void onCompleted() {
                System.out.println("[callSayHello]complete");
            }
        });
    }

    /**
     * 调用sayHelloServerStream
     */
    public void callSayHelloServerStream() {
        //构建请求参数
        HelloRequest helloRequest = HelloRequest.newBuilder()
                .setName("luo")
                .build();

        //阻塞API
        Iterator<HelloReply> helloReplyIterator = this.helloBlockingStub.sayHelloServerStream(helloRequest);
        while (helloReplyIterator.hasNext()) {
            HelloReply helloReply = helloReplyIterator.next();
            System.out.printf("[callSayHelloServerStream]blocking resp: %s", helloReply);
        }


        //非阻塞API
        this.helloStub.sayHelloServerStream(helloRequest, new StreamObserver<HelloReply>() {
            @Override
            public void onNext(HelloReply value) {
                System.out.printf("[callSayHelloServerStream]resp: %s", value);
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("[callSayHelloServerStream]error");
            }

            @Override
            public void onCompleted() {
                System.out.println("[callSayHelloServerStream]complete");
            }
        });
    }

    /**
     * 调用sayHelloClientStream
     */
    public void callSayHelloClientStream() {
        //仅支持非阻塞API
        StreamObserver<HelloRequest> requestObserver = this.helloStub.sayHelloClientStream(new StreamObserver<HelloReply>() {
            @Override
            public void onNext(HelloReply value) {
                System.out.printf("[callSayHelloClientStream]resp: %s", value);
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("[callSayHelloClientStream]error");
            }

            @Override
            public void onCompleted() {
                System.out.println("[callSayHelloClientStream]complete");
            }
        });

        //发送请求
        requestObserver.onNext(HelloRequest.newBuilder().setName("luo1-c").build());
        //连续发送请求
        requestObserver.onNext(HelloRequest.newBuilder().setName("luo2-c").build());
        //连续发送请求
        requestObserver.onNext(HelloRequest.newBuilder().setName("luo3-c").build());

        //结束发送请求
        requestObserver.onCompleted();

    }

    /**
     * 调用sayHelloBiStream
     */
    public void callSayHelloBiStream() {
        //仅支持非阻塞API
        StreamObserver<HelloRequest> requestObserver = this.helloStub.sayHelloBiStream(new StreamObserver<HelloReply>() {
            @Override
            public void onNext(HelloReply value) {
                System.out.printf("[callSayHelloBiStream]resp: %s", value);
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("[callSayHelloBiStream]error");
            }

            @Override
            public void onCompleted() {
                System.out.println("[callSayHelloBiStream]complete");
            }
        });

        //发送请求
        requestObserver.onNext(HelloRequest.newBuilder().setName("luo1-b").build());
        //连续发送请求
        requestObserver.onNext(HelloRequest.newBuilder().setName("luo2-b").build());
        //连续发送请求
        requestObserver.onNext(HelloRequest.newBuilder().setName("luo3-b").build());

        //结束发送请求
        requestObserver.onCompleted();
    }
}

3.6 启动测试

即先启动Server端HelloServer
然后启动Client端HelloClient。

HelloServer控制台输出如下：

[sayHello]recv: name: "luo"
[sayHello]resp: message: "Hello luo"
[sayHello]resp completed!
[sayHello]recv: name: "luo"
[sayHello]resp: message: "Hello luo"
[sayHello]resp completed!
[sayHelloServerStream]recv: name: "luo"
[sayHelloServerStream]resp: message: "Hello_0 luo"
[sayHelloServerStream]resp: message: "Hello_1 luo"
[sayHelloServerStream]resp: message: "Hello_2 luo"
[sayHelloServerStream]resp: message: "Hello_3 luo"
[sayHelloServerStream]resp: message: "Hello_4 luo"
[sayHelloServerStream]resp completed!
[sayHelloServerStream]recv: name: "luo"
[sayHelloServerStream]resp: message: "Hello_0 luo"
[sayHelloServerStream]resp: message: "Hello_1 luo"
[sayHelloServerStream]resp: message: "Hello_2 luo"
[sayHelloServerStream]resp: message: "Hello_3 luo"
[sayHelloServerStream]resp: message: "Hello_4 luo"
[sayHelloServerStream]resp completed!
[sayHelloClientStream]recv_1: luo1-c
[sayHelloClientStream]recv_2: luo2-c
[sayHelloClientStream]recv_3: luo3-c
[sayHelloBiStream]recv: luo1-b
[sayHelloBiStream]resp: message: "Hello luo1-b"
[sayHelloClientStream]resp: message: "Hello luo1-c,luo2-c,luo3-c"
[sayHelloClientStream]resp completed!
[sayHelloBiStream]recv: luo2-b
[sayHelloBiStream]resp: message: "Hello luo2-b"
[sayHelloBiStream]recv: luo3-b
[sayHelloBiStream]resp: message: "Hello luo3-b"
[sayHelloBiStream]resp completed!

HelloClient控制台输出如下：

[callSayHello]blocking resp: message: "Hello luo"
[callSayHello]resp: message: "Hello luo"
[callSayHello]complete
[callSayHelloServerStream]blocking resp: message: "Hello_0 luo"
[callSayHelloServerStream]blocking resp: message: "Hello_1 luo"
[callSayHelloServerStream]blocking resp: message: "Hello_2 luo"
[callSayHelloServerStream]blocking resp: message: "Hello_3 luo"
[callSayHelloServerStream]blocking resp: message: "Hello_4 luo"
[callSayHelloServerStream]resp: message: "Hello_0 luo"
[callSayHelloServerStream]resp: message: "Hello_1 luo"
[callSayHelloClientStream]resp: message: "Hello luo1-c,luo2-c,luo3-c"
[callSayHelloClientStream]complete
[callSayHelloBiStream]resp: message: "Hello luo1-b"
[callSayHelloServerStream]resp: message: "Hello_2 luo"
[callSayHelloBiStream]resp: message: "Hello luo2-b"
[callSayHelloServerStream]resp: message: "Hello_3 luo"
[callSayHelloServerStream]resp: message: "Hello_4 luo"
[callSayHelloServerStream]complete
[callSayHelloBiStream]resp: message: "Hello luo3-b"
[callSayHelloBiStream]complete

参考：
https://www.grpc.io/docs/
https://www.grpc.io/docs/languages/java/
https://github.com/grpc/grpc-java
https://developers.google.cn/protocol-buffers
https://developers.google.cn/protocol-buffers/docs/proto3