【通讯】以json的方式访问grpc

什么是 RPC?

​​RPC​​是​​Remote Procedure Call​​的简称，中文叫远程过程调用。

可以这么理解：现在有两台服务器A和B。部署在A服务器上的应用，想调用部署在B服务器上的另一个应用提供的方法，由于不在一个内存空间，不能直接调用，需要通过网络来达到调用的效果。

现在，我们在A服务的一个本地方法中封装调用B的逻辑，然后只需要在本地使用这个方法，就达到了调用B的效果。

对使用者来说，屏蔽了细节。你只需要知道调用这个方法返回的结果，而无需关注底层逻辑。

那，从封装的那个方法角度来看，调用B之前我们需要知道什么？

当然是一些约定啊。比如，

• 调用的语义，也可以理解为接口规范。(比如​​RESTful​​)
• 网络传输协议 (比如​​HTTP​​)
• 数据序列化反序列化规范(比如​​JSON​​)。

有了这些约定，我就知道如何给你发数据，发什么样的数据，你返回给我的又是什么样的数据。

从上图中可以看出，​​RPC​​​是一种客户端-服务端（​​Client/Server​​）模式。

从某种角度来看，所有本身应用程序之外的调用都可以归类为​​RPC​​​。无论是微服务、第三方​​HTTP​​​接口，还是读写数据库中间件​​Mysql​​​、​​Redis​​。

HTTP 和 RPC 有什么区别？

首先这个问题本身不太严谨。

​​HTTP​​​只是一个通信协议，工作在​​OSI​​第七层。

而​​RPC​​是一个完整的远程调用方案。它包含了:接口规范、传输协议、数据序列化反序列化规范。

这样看，​​RPC​​​和 ​​HTTP​​​的关系只可能是包含关系。为什么是可能？因为​​RPC​​​传输协议那块我可以不基于​​HTTP​​呀。

所以这个问题应该改成:基于​​HTTP​​的远程调用方案 (如:​​HTTP​​+​​RESTful​​+​​JSON​​) 和直接使用RPC远程调用方案有什么区别？

RPC 和 gRPC 有什么关系？

​​gRPC​​是由 ​​google​​开发的一个高性能、通用的开源​​RPC​​框架，主要面向移动应用开发且基于​​HTTP/2​​协议标准而设计，同时支持大多数流行的编程语言。

​​gRPC​​​基于 ​​HTTP/2​​​协议传输。而​​HTTP/2​​​相比​​HTTP1.x​​，有以下一些优势:

用于数据传输的二进制分帧

​​HTTP/2​​​采用二进制格式传输协议，而非​​HTTP/1.x​​的文本格式。

多路复用

​​HTTP/2​​支持通过同一个连接发送多个并发的请求。

而​​HTTP/1.x​​​虽然通过​​pipeline​​也能并发请求，但多个请求之间的响应依然会被阻塞。

服务端推送

服务端推送是一种在客户端请求之前发送数据的机制。在​​HTTP/2​​​中，服务器可以对客户端的一个请求发送多个响应。而不像​​HTTP/1.X​​​一样，只能通过客户端发起​​request​​​,服务端才产生对应的​​response​​。

减少网络流量的头部压缩。

​​HTTP/2​​对消息头进行了压缩传输，能够节省消息头占用的网络流量。至于如何压缩的，可以查看这篇：HPACK: Header Compression for HTTP/2[1]

同时​​gRPC​​​使用​​Protocol Buffers​​​作为序列化协议。关于​​Protocol Buffers​​。官网有一句介绍：

Protocol buffers are Google’s language-neutral, platform-neutral, extensible mechanism for serializing structured data – think XML, but smaller, faster, and simpler.

它是一种与语言、平台无关 、可扩展的序列化结构数据。它的定位类似于​​JSON​​、​​XML​​，但是比他们更小、更快、更简单。

工程演示

新建3个工程

<!--引入的子模块-->
    <modules>
        <module>grpc-lib</module>
        <module>grpc-server</module>
        <module>grpc-client</module>
    </modules>

服务端

@Slf4j
@GrpcService(GreeterOuterClass.class)
public class GreeterService extends GreeterGrpc.GreeterImplBase {
    @Override
    public void sayHello(GreeterOuterClass.HelloRequest request, StreamObserver<GreeterOuterClass.HelloReply> responseObserver) {
        String message = "Hello " + request.getName();
        final GreeterOuterClass.HelloReply.Builder replyBuilder = GreeterOuterClass.HelloReply.newBuilder().setMessage(message);
        responseObserver.onNext(replyBuilder.build());
        responseObserver.onCompleted();
        log.info("Returning " + message);
    }
}

客户端

@Service
public class GrpcClientService {


    //两种获取Channel方式
    //方法一
    @GrpcClient("local-grpc-server")
    private Channel serverChannel;


    public String sendMessage(String name) {
        GreeterGrpc.GreeterBlockingStub stub = GreeterGrpc.newBlockingStub(serverChannel);
        GreeterOuterClass.HelloReply response = stub.sayHello(GreeterOuterClass.HelloRequest.newBuilder().setName(name).build());
        return response.getMessage();
    }


    //方法二
    @Value("${grpc.client.local-grpc-server.port}")
    private Integer port;


    private String host = "0.0.0.0";


    public List<InterruptStatusVo> getResult(String building, String park, String region) {
        //通过端口号去获取
        io.grpc.Channel channel = NettyChannelBuilder.forAddress(host, port)
                .negotiationType(NegotiationType.PLAINTEXT)
                .build();


        InterruptGrpc.InterruptBlockingStub stub  =  InterruptGrpc.newBlockingStub(channel);
        InterruptStatus.RequestParam.Builder param = InterruptStatus.RequestParam.newBuilder();
        if(Objects.nonNull(building)){
            param.setLocationBuilding(building);
        }
        if(Objects.nonNull(park)){
            param.setLocationPark(park);
        }
        if(Objects.nonNull(region)){
            param.setLocationRegion(region);
        }


        InterruptStatus.Message response = stub.getInterruptStatusData(param.build());


        List<InterruptStatusVo> datalist = new ArrayList<>();
        List<InterruptStatus.data>  data = response.getDatalistList();
        for(InterruptStatus.data data1: data){
            InterruptStatusVo interruptStatusVo = new InterruptStatusVo();
            BeanUtils.copyProperties(data1, interruptStatusVo);
            List<InterruptStatus.data.interruptProject> interruptProjects = data1.getInterruptProjectsList();
            List<InterruptProjectVo> interruptProjectVoList = new ArrayList<>();
            for(InterruptStatus.data.interruptProject project: interruptProjects){
                InterruptProjectVo interruptProjectVo = new InterruptProjectVo();
                interruptProjectVo.setFloors(project.getFloors());
                com.google.protobuf.ProtocolStringList strings = project.getProjectList();


                int count = strings.size();
                String[] projects = new String[count];
                for(int i = 0; i < count; i++){
                    projects[i] = strings.get(i);
                }
                interruptProjectVo.setProject(projects);
                interruptProjectVoList.add(interruptProjectVo);
            }
            interruptStatusVo.setInterruptProjects(interruptProjectVoList);
            datalist.add(interruptStatusVo);
        }
        return  datalist;
    }








}

测试

@RestController
public class GrpcClientController {


    @Autowired
    private GrpcClientService grpcClientService;


    @RequestMapping("/")
    public String printMessage(@RequestParam(defaultValue = "sf") String name) {
        return grpcClientService.sendMessage(name);
    }


    @RequestMapping("/test")
    public Object printMessage(@RequestParam(defaultValue = "中关村") String region,
                               @RequestParam(defaultValue = "A区") String park,
                               @RequestParam(defaultValue = "A01") String building) {
        return grpcClientService.getResult(null, null, null);
    }
}

分别启动服务端和客户端。

项目源码

链接: https://pan.baidu.com/s/10F-bLs7h4NC52HjDVWlaBA 提取码: zftp

Protocol buffer协议介绍

Google Protocol Buffer( 简称 Protobuf) 是 Google 公司内部的混合语言数据标准，用于 RPC 系统和持续数据存储系统。

Protocol Buffers 是一种轻便高效的结构化数据存储格式，可以用于结构化数据串行化，或者说序列化。它很适合做数据存储或 RPC 数据交换格式。可用于通讯协议、数据存储等领域的语言无关、平台无关、可扩展的序列化结构数据格式。目前提供了 C++、Java、Python、ruby、php、go等多种语言的 API。

Message格式介绍：

在.proto文件定义消息，message是.proto文件最小的逻辑单元，由一系列name-value键值对构成。

消息由至少一个字段组合而成，类似于C语言中的结构。每个字段都有一定的格式
字段格式：限定修饰符① | 数据类型② | 字段名称③ | = | 字段编码值④ | [字段默认值⑤]

限定修饰符包含：required optional repeated

数据类型包含：bool、int32、float、string、bytes、enum、message等

字段名称：建议字段的命名采用以下划线分割的驼峰式。例如 first_name 而不是firstName.

字段编码值：有了该值，通信双方才能互相识别对方的字段。编码值的取值范围为 1~2^32（4294967296），1900~2000编码值为Google protobuf 系统内部保留值。

字段默认值：是可选字段，代表该字段的默认值

Pb的编码实例：

现在举一个例子说明protocol buffer的传输机制，先看一个最简单的messgae定义：

message test{

  required int32 a=1 [default = 150]；

}

将该对象序列化到二进制文件中，可以看到文本中的数据是：

08 96 01

在理解Protocol Buffer的编码规则之前，首先需要了解varints: varints是一种使用一个或多个字节表示整型数据的方法。其中数值本身越小，其所占用的字节数越少。

Protocol buffer对整数的编码优化：
      在varint中，除了最后一个字节之外的每个字节中都包含一个msb(most significant bit)设置(使用最高位)，这意味着其后的字节是否和当前字节一起来表示同一个整型数值。而字节中的其余七位将用于存储数据本身。由此我们可以简单的解释一下Base 128，通常而言，整数数值都是由字节表示，其中每个字节为8位，即Base 256。然而在Protocol Buffer的编码中，最高位成为了msb，只有后面的7位存储实际的数据，因此我们称其为Base 128（2的7次方）。

首先先看96 01 ，二进制是1001 0110 0000 0001，根据varint的规则：

  1001 0110 0000 0001

à  001 0110 0000 0001 #去掉msb位，此时msb是1，代表下一个字节和当前字节一起表示某个整数

à  0000 0001 001 0110  #字节序转换

à  1001 0110

à  128 + 16 + 4 + 2 = 150

再来看08的解码过程，在进行消息编码时，key/value被连接成字节流，value的值是150已经解码完成，08就是key的值，key的值根据protocol定义是由字段编码值和字段数据类型合成编码所得，公式如下：

Key = field_number << 3 | field_type

Protocol Buffer可以支持的字段类型：

定义的字段数据类型是int32，属于varint，对应的type是0，定义的字段编码值是1，所以：

08 = 1 << 3 | 0

Protocol buffer对字符串的编码优化：

但是如果Value是一个很长的字符串，每个字节都拿出1个比特来区分边界就太浪费空间了，而且字符串本身就是一个一个字节的，被打乱后也会影响解码效率。因此，PB将Value长度信息的指示可以放在Key和Value之间。（长度本身也是一个整数，就用前面那种方法进行编码即可），在解码Value时，解析长度就可以知道Value值到哪里结束。

这样就完成了对test的编码和解码。

代码实例解析：

Message定义

1. 该实例代码主要实现了hello world得回显功能，首先定义example.proto文件，文件内容如下：

syntax = "proto3";




option java_package = "com.lrbj.grpc.lib";


// The greeting service definition.
service Greeter{
    // Sends a greeting
    rpc SayHello (HelloRequest) returns (HelloReply) {
    }
}


// The request message containing the user's name.
message HelloRequest {
    string name = 1;
}


// The response message containing the greetings
message HelloReply {
    string message = 1;
}

第一行指定了使用proto3语法，如果没有指定，编译器会使用proto2。这个指定语法行必须是文件的非空非注释的第一个行。

如果想要将消息类型用在GRPC系统中，可以在.proto文件中定义一个RPC服务接口，protocol buffer编译器将会根据所选择的不同语言生成服务接口代码及存根。

grpc允许定义4种类型的 service 方法：简单rpc，服务器端流式rpc，客户端流式rpc，双向流式rpc，本实例中Search方法只是一个简单的rpc，不涉及到流式操作。

gRPC默认的序列化方式是protobuf.。

Windows下使用，需要下载，下载地址如下：

​​https://github.com/protocolbuffers/protobuf/releases​​

下载后解压，查看bin下的protoc文件:

将bin文件夹下protoc应用程序复制到C:\Program Files\Go\bin:

执行

protoc --version

SpringCloud GateWay

官方文档：

​​https://docs.spring.io/spring-cloud-gateway/docs/current/reference/html/#gateway-starter​​

Spring Cloud Gateway 是Spring Cloud的一个全新的API网关项目，目的是为了替换掉Zuul1，它基于Spring5.0 + SpringBoot2.0 + WebFlux（基于性能的Reactor模式响应式通信框架Netty，异步阻塞模型）等技术开发，性能于Zuul，官测试，Spring Cloud GateWay是Zuul的1.6倍 ，旨在为微服务架构提供种简单有效的统的API路由管理式。

可以与Spring Cloud Discovery Client（如Eureka）、Ribbon、Hystrix等组件配合使用，实现路由转发、负载均衡、熔断、鉴权、路径重写、志监控等

1. Gateway还内置了限流过滤器，实现了限流的功能。
2. 设计优雅，容易拓展

基本概念

路由（Route）是GateWay中最基本的组件之一，表示一个具体的路由信息载体，主要由下面几个部分组成：

1. id：路由唯一标识，区别于其他的route
2. url：路由指向的目的地URL，客户端请求最终被转发到的微服务
3. order：用于多个Route之间的排序，数值越小越靠前，匹配优先级越高
4. predicate：断言的作用是进行条件判断，只有断言为true，才执行路由
5. filter: 过滤器用于修改请求和响应信息
6. 
   

核心流程

核心概念：

1. ​​Gateway Client​​​ 向 ​​Spring Cloud Gateway​​ 发送请求
2. 请求首先会被 ​​HttpWebHandlerAdapter​​ 进行提取组装成网关上下文
3. 然后网关的上下文会传递到 ​​DispatcherHandler​​ ，它负责将请求分发给 ​​RoutePredicateHandlerMapping​​
4. ​​RoutePredicateHandlerMapping​​ 负责路由查找，并根据路由断言判断路由是否可用
5. 如果过断言成功，由​​FilteringWebHandler​​ 创建过滤器链并调用
6. 通过特定于请求的 ​​Fliter​​ 链运行请求，​​Filter​​ 被虚线分隔的原因是Filter可以在发送代理请求之前（pre）和之后（post）运行逻辑
7. 执行所有pre过滤器逻辑。然后进行代理请求。发出代理请求后，将运行“post”过滤器逻辑。
8. 处理完毕之后将 ​​Response​​ 返回到 ​​Gateway​​ 客户端

Filter过滤器：

• Filter在pre类型的过滤器可以做参数效验、权限效验、流量监控、日志输出、协议转换等。
• Filter在post类型的过滤器可以做响应内容、响应头的修改、日志输出、流量监控等

核心思想

当用户发出请求达到 ​​GateWay​​ 之后，会通过一些匹配条件，定位到真正的服务节点，并且在这个转发过程前后，进行一些细粒度的控制，其中 Predicate（断言） 是我们的匹配条件，Filter 是一个拦截器，有了这两点，再加上URL，就可以实现一个具体的路由，核心思想：路由转发+执行过滤器链

这个过程就好比考试，我们考试首先要找到对应的考场，我们需要知道考场的地址和名称(id和url)，然后我们进入考场之前会有考官查看我们的准考证是否匹配（断言），如果匹配才会进入考场，我们进入考场之后，（路由之前）会进行身份的登记和考试的科目，填写考试信息，当我们考试完成之后（路由之后）会进行签字交卷，走出考场，这个就类似我们的过滤器

Route(路由) ：构建网关的基础模块，由ID、目标URL、过滤器等组成

Predicate（断言） ：开发人员可以匹配HTTP请求中的内容（请求头和请求参数），如果请求断言匹配贼进行路由

Filter（过滤） ：GateWayFilter的实例，使用过滤器，可以在请求被路由之前或者之后对请求进行修改

整合网关和grpc

这个功能与上面的基本相同，只是加上了http2的安全加密功能。

hello.proto定义如下：

syntax = "proto3";
option java_multiple_files = true;
package com.example.grpcserver.hello;


message HelloRequest {
  string firstName = 1;
  string lastName = 2;
}


message HelloResponse {
  string greeting = 1;
}


service HelloService {
  rpc hello(HelloRequest) returns (HelloResponse);
}

服务的启动顺序如下：

先启动grpc-server

./gradlew :grpc-server:bootRun

然后启动simple-gateway

./gradlew :grpc-simple-gateway:bootRun

最后启动客户端测试。

./gradlew :grpc-client:bootRun

工程的结构如图：

英文的demo地址如下：

https://github.com/Albertoimpl/spring-cloud-gateway-grpc

我已经在这个的基础之上进行了修改，能够咋国内直接下载相关的jar包。

以json的方式访问grpc

使用的是内置的JSONToGRPCFilter ，在json提交之后，将json请求转发到grpc-server，最主要的转换类如下：

static class GRPCResponseDecorator extends ServerHttpResponseDecorator {


  @Override
  public Mono<Void> writeWith(Publisher<?extends DataBuffer> body) {
    exchange.getResponse().getHeaders().set("Content-Type", "application/json");


    URI requestURI = exchange.getRequest().getURI();
    ManagedChannel channel = createSecuredChannel(requestURI.getHost(), 6565);


    return getDelegate().writeWith(deserializeJSONRequest()
            .map(jsonRequest -> {
                String firstName = jsonRequest.getFirstName();
                String lastName = jsonRequest.getLastName();
                return HelloServiceGrpc.newBlockingStub(channel)
                        .hello(HelloRequest.newBuilder()
                                .setFirstName(firstName)
                                .setLastName(lastName)
                                .build());
            })
            .map(this::serialiseJSONResponse)
            .map(wrapGRPCResponse())
            .cast(DataBuffer.class)
            .last());
  }
}

完整的实现

grpc-json-gateway/src/main/java/com/example/grpcserver/hello/JSONToGRPCFilterFactory.java

启动grpc-server，然后启动grpc-json-gateway，然后发送请求即可。

在发送json请求的时候需要主要，如果用postman发送，需要进行安全设置。

进入postman的设置界面；

关闭general下面的ssl校验；

然后添加秘钥信息进行安全发送

传入访问的JSON

{"firstName":"Feng","lastName":"San"}

设置header为Content-Type: application/json

编译之后的源码低地址如下：

链接: https://pan.baidu.com/s/1it5dgoygcJvU30y_f1Whzg 提取码: z3d8

​