gRPC快速入门-蒲公英云

gRPC 是 Google 开源的基于 Protobuf 和 Http2.0 协议的通信框架，底层由netty提供。之前也简单的介绍过 HTTP/2 重要特性，gRPC提供四种模式：unary，client streaming，server streaming 以及 bidirectional streaming，对于底层 HTTP/2 来说，它们都是 stream，并且仍然是一套 request + response 模型。

采用gRPC作为底层通讯的开源产品也比较多，比如： Tensorflow、CoreOS、Tidb。

gRPC 特色之处在于它的客户端是多路复用的线程安全的，可以拿过来直接使用，而且多路复用的客户端在效率上有明显优势。

插播句

gRPC 出来的时间要比 Thrift （Facebook ）要晚，Google 自然是知道 gRPC 要比 Thrift 相比有明显的优势所以才敢放出来开源的，在开源影响力上Google 一直很强势

Protobuf

gRPC 的 service 接口是基于 protobuf 定义的，我们可以非常方便的将 service 与 HTTP/2 关联起来。

思考：我们知道Protobuf 协议仅仅定义了单个消息的序列化格式，HTTP/2模式下该如何protobuf集成呢？

难道我们需要自定义头部的长度和消息的类名称来区分消息边界？

Protobuf 将消息序列化后，内容放在 HTTP2.0 的 Data 帧的 Payload 字段里，消息的长度在 Data 帧的 Length 字段里，消息的类名称放在 Headers 帧的 Path 头信息中。我们需要通过头部字段的 Path 来确定具体消息的解码器来反序列化 Data 帧的 Payload 内容，而这些工作 gRPC 都已经封装好了。

Path : /Service-Name/{method name}
Service-Name : ?( {proto package name} “.” ) {service name}
Message-Type : {fully qualified proto message name}
Content-Type : “application/grpc+proto”

其实request-response的奥秘都在这张图中，下面将展开讲

Request

request 通常包含：

Request-Headers，直接使用的 HTTP/2 headers，在 Headers Frame 里面派发，定义的 header 主要有 Call-Definition 以及 Custom-Metadata
Length-Prefixed-Message，主要在 Data Frame 里面派发，它有一个 Compressed flag ，然后后面跟着四字节的 message length 以及实际的 message
EOS（end-of-stream），在最后的 DATA frame 里面带上了 END_STREAM 这个 flag。用来表示 stream 不会在发送任何数据，可以关闭了

Call-Definition 里面包括 Method（其实就是用的 HTTP/2 的 POST），Content-Type 等。

Custom-Metadata 则是应用层自定义的任意 key-value，key 不建议使用 grpc- 开头，因为这是为 gRPC 后续自己保留的。

Compressed flag 用来表示改 message 是否压缩，如果为 1，表示该 message 采用了压缩，而压缩算啊定义在 header 里面的 Message-Encoding 里面。

思考：Request-Header何时算响应结束呢？

上图也有答案，Headers Frame中带上了END_HEADERS 表示已经是头信息的最后一个帧

Response

主要包含：

Response-Headers，主要包括 HTTP-Status，Content-Type 以及 Custom-Metadata 等
Length-Prefixed-Message，
Trailers，包括了 Status 以及 0 或者多个 Custom-Metadata
Trailers-Only，如果遇到了错误，也可以直接返回。也有 HTTP-Status ，Content-Type 和 Trailers

HTTP-Status 就是通常的 HTTP 200，301，400 这些不再解释

Status 也就是 gRPC 的 status

Status-Message 则是 gRPC 的 message，采用了 Percent-Encoded 的编码方式，具体参考这里

思考：response何时算响应结束呢？

从上图就可以找到答案，最后收到的 Headers frame 里面，带上了 Trailers，并且DATA frame有 END_STREAM 这个 flag，那么就意味着 response结束！

接下来将带你从实操中理解gRPC的一些抽象的概念，可下载完整的源码。

os-maven-plugin

它主要是用于根据不同操作系统提供不同的变量（protoc版本信息）

[INFO] os.detected.name: windows
[INFO] os.detected.arch: x86_64
[INFO] os.detected.version: 6.1
[INFO] os.detected.version.major: 6
[INFO] os.detected.version.minor: 1
[INFO] os.detected.classifier: windows-x86_64

maven-dependency-plugin

自动下载protoc（可执行程序）到本地目录

protobuf-maven-plugin

针对*.proto文件生成java文件（主要是contract，dto）

服务提供方

gRPC 的一个特色之处在于提供了 Streaming 模式，有了 Streaming 模式，客户端可以将一连串的请求连续发送到服务器，服务器也可以将一连串连续的响应回复给客户端。Streaming 就是双工对话，它允许一个人同时说又同时听，如果你对rxjava有一定的了解的话，我擦，是不是有种熟悉的感觉！

public class OrderServiceProvider extends OrderServiceGrpc.OrderServiceImplBase{
    @Override
    public void getOrder(OrderId request, StreamObserver<OrderDTO> responseObserver) {
        responseObserver.onNext(OrderDTO.newBuilder().setName("test").build());
        responseObserver.onCompleted();
    }
    @Override
    public void addOrder(OrderDTO request, StreamObserver<OrderId> responseObserver) {
        responseObserver.onNext(OrderId.newBuilder().setId(9527).build());
        responseObserver.onCompleted();
    }
    @Override
    public void queryOrder(OrderDTO request, StreamObserver<OrderId> responseObserver) {
        responseObserver.onNext(OrderId.newBuilder().setId(9527).build());
        responseObserver.onNext(OrderId.newBuilder().setId(9528).build());
        responseObserver.onCompleted();
    }
}

客户消费端

gRPC提供了丰富的Contract(客户端Sevice协议)的实现方式，google的解决方案显然很棒，不像dubbo比较凑合。

回顾下dubbo

在2.7版本之前：

将同步接口声明成 async=true，比如<dubbo:reference id="asyncService" interface="xxxx" async="true"/>
通过上下文类获取 future，比如：Future fooFuture = RpcContext.getContext().getFuture();

问题：不太符合异步编程的习惯，如果同时进行多个异步调用，使用不当很容易造成上下文污染，jdk原生Future 并不支持 callback 的调用方式。

在2.7.1（孵化中的正式版本）中改造：显式声明异步接口使用了JDK1.8 的CompletableFuture
//声明
public interface AsyncService {
    String sayHello(String name);
    default CompletableFuture<String> sayHiAsync(String name) {
        return CompletableFuture.completedFuture(sayHello(name));
    }
}
//调用
CompletableFuture<String> future = asyncService.sayHiAsync("Han MeiMei");
future.whenComplete((retValue, exception) -> {
    if (exception == null) {
        System.out.println(retValue);
    } else {
        exception.printStackTrace();
    }
});
而这个异步更多是客户端异步，虽然2.7 新增了服务端异步的支持，然而个人认为服务端异步的特性较为鸡肋

but，dubbo在3.0.0-SNAPSHOT中已经有了彻底的改变，支持响应式之RSocket

提供了3种Contract：

BlockingStub，阻塞式
FutureStub，guava的ListenableFuture

Stub，支持StreamObserver

public class OrderClient {

private String host = "localhost";
private int serverPort = 9527;
private ManagedChannel managedChannel;
private OrderServiceGrpc.OrderServiceBlockingStub orderConsumer;
private OrderServiceGrpc.OrderServiceFutureStub orderAsyncConsumer;
private OrderServiceGrpc.OrderServiceStub orderStreamConsumer;
private OrderId orderDTO = OrderId.newBuilder().setId(123).build();
@Before
public void init() {
    managedChannel = ManagedChannelBuilder.forAddress(host, serverPort)
            //Channels are secure by default (via SSL/TLS)
            .usePlaintext().
                    build();
    orderConsumer = OrderServiceGrpc.newBlockingStub(managedChannel);
    orderAsyncConsumer = OrderServiceGrpc.newFutureStub(managedChannel);
    orderStreamConsumer = OrderServiceGrpc.newStub(managedChannel);
}
@Test
public void getOrder() {
    OrderDTO result = orderConsumer.getOrder(orderDTO);
    System.out.println(result);
}
@Test
public void asyncGetOrder() {
    ListenableFuture<OrderDTO> result = orderAsyncConsumer.getOrder(orderDTO);
    System.out.println(result);
}
@Test
public void streamGetOrder() throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(1);
    StreamObserver<OrderDTO> responseObserver = new StreamObserver<OrderDTO>() {
        @Override
        public void onNext(OrderDTO value) {
            System.out.println("get result :" + value);
        }
        @Override
        public void onError(Throwable t) {
            Status status = Status.fromThrowable(t);
            System.out.println("failed with status : " + status);
            latch.countDown();
        }
        @Override
        public void onCompleted() {
            System.out.println("finished!");
            latch.countDown();
        }
    };
    orderStreamConsumer.getOrder(orderDTO, responseObserver);
    latch.await();
}