Dubbo 在跨语言和协议穿透性方向上的探索：支持 HTTP/2 gRPC 和 Protobuf

基本介绍

Dubbo 协议

• Dubbo 协议体 Body 中有一个可扩展的 attachments 部分，这给 RPC 方法之外额外传递附加属性提供了可能，是一个很好的设计。但是类似的 Header 部分，却缺少类似的可扩展 attachments，这点可参考 HTTP 定义的 Ascii Header 设计，将 Body Attachments 和 Header Attachments 做职责划分。
• Body 协议体中的一些 RPC 请求定位符如 Service Name、Method Name、Version 等，可以提到 Header 中，和具体的序列化协议解耦，以更好的被网络基础设施识别或用于流量管控。
• 扩展性不够好，欠缺协议升级方面的设计，如 Header 头中没有预留的状态标识位，或者像 HTTP 有专为协议升级或协商设计的特殊 packet。
• 在 Java 版本的代码实现上，不够精简和通用。如在链路传输中，存在一些语言绑定的内容；消息体中存在冗余内容，如 Service Name 在 Body 和 Attachments 中都存在。

HTTP/1

1. HTTP 的语义和可扩展性能很好的满足 RPC 调用需求。
2. 通用性，HTTP 协议几乎被网络上的所有设备所支持，具有很好的协议穿透性。

• 典型的 Request – Response 模型，一个链路上一次只能有一个等待的 Request 请求
• HTTP/1 支持 Keep-Alive 链接，避免了链接重复创建开销
• Human Readable Headers，使用更通用、更易于人类阅读的头部传输格式
• 无直接 Server Push 支持，需要使用 Polling Long-Polling 等变通模式

HTTP/2

• 支持单条链路上的 Multiplexing，相比于 Request - Response 独占链路，基于 Frame 实现更高效利用链路
• Request - Stream 语义，原生支持 Server Push 和 Stream 数据传输
• Flow Control，单条 Stream 粒度的和整个链路粒度的流量控制
• 头部压缩 HPACK
• Binary Frame
• 原生 TLS 支持

gRPC

• Coverage & Simplicity，协议设计和框架实现要足够通用和简单，能运行在任何设备之上，甚至一些资源首先的如 IoT、Mobile 等设备。
• Interoperability & Reach，要构建在更通用的协议之上，协议本身要能被网络上几乎所有的基础设施所支持。
• General Purpose & Performant，要在场景和性能间做好平衡，首先协议本身要是适用于各种场景的，同时也要尽量有高的性能。
• Payload Agnostic，协议上传输的负载要保持语言和平台中立。
• Streaming，要支持 Request - Response、Request - Stream、Bi-Steam 等通信模型。
• Flow Control，协议自身具备流量感知和限制的能力。
• Metadata Exchange，在 RPC 服务定义之外，提供额外附加数据传输的能力。

Protobuf

• 跨语言 跨平台，语言中立的数据描述格式，默认提供了生成多种语言的 Compiler 工具。
• 安全性，由于反序列化的范围和输出内容格式都是 Compiler 在编译时预生成的，因此绕过了类似 Java Deserialization Vulnarability 的问题。
• 二进制 高性能
• 强类型
• 字段变更向后兼容

message Person {
      required string name = 1;
      required int32 id = 2;
      optional string email = 3;
      enum PhoneType {
        MOBILE = 0;
        HOME = 1;
        WORK = 2;
      }
      message PhoneNumber {
        required string number = 1;
        optional PhoneType type = 2 [default = HOME];
      }
      repeated PhoneNumber phone = 4;
    }

service SearchService {
    rpc Search (SearchRequest) returns (SearchResponse);
}

Dubbo 所做的支持

• 协议，上面我们已经分析过 Dubbo 协议既有的缺点，如果能在 HTTP/2 之上构建应用层协议，则无疑能避免这些弊端，同时最大可能的提高协议的穿透性，避免网关等协议转换组件的存在，更有利于链路上的流量管控。考虑到 gRPC 是构建在 HTTP/2 之上，并且已经是云原生领域推荐的通信协议，Dubbo 在第一阶段选择了直接支持 gRPC 协议作为当前的 HTTP/2 解决方案。我们也知道 gRPC 框架自身的弊端在于易用性不足以及服务治理能力欠缺（这也是目前绝大多数厂商不会直接裸用 gRPC 框架的原因），通过将其集成进 Dubbo 框架，用户可以方便的使用 Dubbo 编程模型 + Dubbo 服务治理 + gRPC 协议通信的组合。
• 服务定义，当前 Dubbo 的服务定义和具体的编程语言绑定，没有提供一种语言中立的服务描述格式，比如 Java 就是定义 Interface 接口，到了其他语言又得重新以另外的格式定义一遍。因此 Dubbo 通过支持 Protobuf 实现了语言中立的服务定义。
• 序列化，Dubbo 当前支持的序列化包括 Json、Hessian2、Kryo、FST、Java 等，而这其中支持跨语言的只有 Json、Hessian2，通用的 Json 有固有的性能问题，而 Hessian2 无论在效率还是多语言 SDK 方面都有所欠缺。为此，Dubbo 通过支持 Protobuf 序列化来提供更高效、易用的跨语言序列化方案。

示例

示例 1，使用 Dubbo 开发 gRPC 服务

1. 定义服务 IDL

syntax = "proto3";
    
    option java_multiple_files = true;
    option java_package = "io.grpc.examples.helloworld";
    option java_outer_classname = "HelloWorldProto";
    option objc_class_prefix = "HLW";
    
    package helloworld;
    
    // 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;
    }

2. Protobuf Compiler 生成 Stub

1. 定义 Maven Protobuf Compiler 插件工具。这里我们扩展了 Protobuf 的 Compiler 工具，以用来生成 Dubbo 特有的 RPC stub，此当前以 Maven 插件的形式发布。

<plugin>
  <groupId>org.xolstice.maven.plugins</groupId>
  <artifactId>protobuf-maven-plugin</artifactId>
  <version>0.5.1</version>
  <configuration>
    <protocArtifact>com.google.protobuf:protoc:3.7.1:exe:${os.detected.classifier}    
    </protocArtifact>
    <pluginId>dubbo-grpc-java</pluginId>
    <pluginArtifact>org.apache.dubbo:protoc-gen-dubbo-java:1.19.0-SNAPSHOT:exe:${os.detected.classifier}</pluginArtifact>
    <outputDirectory>build/generated/source/proto/main/java</outputDirectory>
    <clearOutputDirectory>false</clearOutputDirectory>
    <pluginParameter>grpc</pluginParameter>
  </configuration>
  <executions>
    <execution>
      <goals>
        <goal>compile</goal>
        <goal>compile-custom</goal>
      </goals>
    </execution>
  </executions>
</plugin>

<pluginArtifact>org.apache.dubbo:protoc-gen-dubbo-java:1.19.0-SNAPSHOT:exe:${os.detected.classifier}</pluginArtifact>

<pluginParameter>grpc</pluginParameter>

2. 生成 Java Bean 和 Dubbo-gRPC stub

# 运行以下 maven 命令
$ mvn clean compile

/**
* Code generated for Dubbo
*/
public interface IGreeter {
  default public io.grpc.examples.helloworld.HelloReply     sayHello(io.grpc.examples.helloworld.HelloRequest request) {
    throw new UnsupportedOperationException("No need to override this method, extend XxxImplBase and override all methods it allows.");
  }
  default public com.google.common.util.concurrent.ListenableFuture<io.grpc.examples.helloworld.HelloReply> sayHelloAsync(
    io.grpc.examples.helloworld.HelloRequest request) {
    throw new UnsupportedOperationException("No need to override this method, extend XxxImplBase and override all methods it allows.");
  }
  public void sayHello(io.grpc.examples.helloworld.HelloRequest request,
                       io.grpc.stub.StreamObserver<io.grpc.examples.helloworld.HelloReply> responseObserver);
}

3. 业务逻辑开发

package org.apache.dubbo.samples.basic.impl;
import io.grpc.examples.helloworld.GreeterGrpc;
import io.grpc.examples.helloworld.HelloReply;
import io.grpc.examples.helloworld.HelloRequest;
import io.grpc.stub.StreamObserver;
public class GrpcGreeterImpl extends GreeterGrpc.GreeterImplBase {
  @Override
  public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver)         {
    System.out.println("Received request from client.");
    System.out.println("Executing thread is " + Thread.currentThread().getName());
    HelloReply reply = HelloReply.newBuilder()
      .setMessage("Hello " +     request.getName()).build();
    responseObserver.onNext(reply);
    responseObserver.onCompleted();
  }
}

1. Provider 端暴露 Dubbo 服务

<dubbo:application name="demo-provider"/>
<!-- 指定服务暴露协议为 gRPC -->
<dubbo:protocol id="grpc" name="grpc"/>
<dubbo:registry address="zookeeper://${zookeeper.address:127.0.0.1}:2181"/>
<bean id="greeter" class="org.apache.dubbo.samples.basic.impl.GrpcGreeterImpl"/>
<!-- 指定 protoc-gen-dubbo-java 生成的接口 -->
<dubbo:service interface="io.grpc.examples.helloworld.GreeterGrpc$IGreeter" ref="greeter" protocol="grpc"/>public static void main(String[] args) throws Exception {
  ClassPathXmlApplicationContext context =
    new ClassPathXmlApplicationContext("spring/dubbo-demo-provider.xml");
  context.start();
  System.out.println("dubbo service started");
  new CountDownLatch(1).await();
}

1. 引用 Dubbo 服务

<dubbo:application name="demo-consumer"/>
<dubbo:registry address="zookeeper://${zookeeper.address:127.0.0.1}:2181"/>
<!-- 指定 protoc-gen-dubbo-java 生成的接口 -->
<dubbo:reference id="greeter" interface="io.grpc.examples.helloworld.GreeterGrpc$IGreeter" protocol="grpc"/>public static void main(String[] args) throws IOException {
  ClassPathXmlApplicationContext context =
    new ClassPathXmlApplicationContext("spring/dubbo-demo-consumer.xml");
  context.start();
  GreeterGrpc.IGreeter greeter = (GreeterGrpc.IGreeter) context.getBean("greeter");
  HelloReply reply = greeter.sayHello(HelloRequest.newBuilder().setName("world!").build());
  System.out.println("Result: " + reply.getMessage());
  System.in.read();
}

示例1附：高级用法

/**
* Code generated for Dubbo
*/
public interface IGreeter {
  default public HelloReply sayHello(HelloRequest request) {
    // ......
  }
  default public ListenableFuture<HelloReply> sayHelloAsync(HelloRequest request) {
    // ......
  }
  public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver);
}

public static void main(String[] args) throws IOException {
  // ...
  GreeterGrpc.IGreeter greeter = (GreeterGrpc.IGreeter) context.getBean("greeter");
  ListenableFuture<HelloReply> future =   
    greeter.sayHAsyncello(HelloRequest.newBuilder().setName("world!").build());
  // ...
}

• org.apache.dubbo.rpc.protocol.grpc.interceptors.ClientInterceptor
• org.apache.dubbo.rpc.protocol.grpc.interceptors.GrpcConfigurator
• org.apache.dubbo.rpc.protocol.grpc.interceptors.ServerInterceptor
• org.apache.dubbo.rpc.protocol.grpc.interceptors.ServerTransportFilter

public interface GrpcConfigurator {
  // 用来定制 gRPC NettyServerBuilder
  default NettyServerBuilder configureServerBuilder(NettyServerBuilder builder, URL url) {
    return builder;
  }
  // 用来定制 gRPC NettyChannelBuilder
  default NettyChannelBuilder configureChannelBuilder(NettyChannelBuilder builder, URL url) {
    return builder;
  }
  // 用来定制 gRPC CallOptions, 定义某个服务在每次请求间传递数据
  default CallOptions configureCallOptions(CallOptions options, URL url) {
    return options;
  }
}

public class MyGrpcConfigurator implements GrpcConfigurator {
  private final ExecutorService executor = Executors
    .newFixedThreadPool(200, new NamedThreadFactory("Customized-grpc", true));
  @Override
  public NettyServerBuilder configureServerBuilder(NettyServerBuilder builder, URL url) {
    return builder.executor(executor);
  }
  @Override
  public NettyChannelBuilder configureChannelBuilder(NettyChannelBuilder builder, URL url)
  {
    return builder.flowControlWindow(10);
  }
  @Override
  public CallOptions configureCallOptions(CallOptions options, URL url) {
    return options.withOption(CallOptions.Key.create("key"), "value");
  }
}

default=org.apache.dubbo.samples.basic.comtomize.MyGrpcConfigurator

1. 指定 Provider 端线程池 默认用的是 Dubbo 的线程池，有 fixed (默认)、cached、direct 等类型。以下演示了切换为业务自定义线程池。

private final ExecutorService executor = Executors
             .newFixedThreadPool(200, new NamedThreadFactory("Customized-grpc", true));
public NettyServerBuilder configureServerBuilder(NettyServerBuilder builder, URL url) 
{
  return builder.executor(executor);
}

1. 指定 Consumer 端限流值 设置 Consumer 限流值为 10

@Override
public NettyChannelBuilder configureChannelBuilder(NettyChannelBuilder builder, URL url)
{
  return builder.flowControlWindow(10);
}

1. 传递附加参数 DemoService 服务调用传递 key

@Override
public CallOptions configureCallOptions(CallOptions options, URL url) {
  if (url.getServiceInterface().equals("xxx.DemoService")) {
    return options.withOption(CallOptions.Key.create("key"), "value");
  } else {
    return options;
  }
}

* MyClientStreamInterceptor，工作在 client 端，拦截发出的请求流和接收的响应流
* MyServerStreamInterceptor，工作在 server 端，拦截收到的请求流和发出的响应流

示例 2， 使用 Protobuf 开发 Dubbo 服务

1. 定义服务

syntax = "proto3";
    option java_multiple_files = true;
    option java_package = "org.apache.dubbo.demo";
    option java_outer_classname = "DemoServiceProto";
    option objc_class_prefix = "DEMOSRV";
    package demoservice;
    // The demo service definition.
    service DemoService {
      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;
    }

2. Compiler 编译服务

1. 引入 Protobuf Compiler Maven 插件，同时指定 protoc-gen-dubbo-java RPC 扩展

<plugin>
  <groupId>org.xolstice.maven.plugins</groupId>
  <artifactId>protobuf-maven-plugin</artifactId>
  <version>0.5.1</version>
  <configuration>
    <protocArtifact>com.google.protobuf:protoc:3.7.1:exe:${os.detected.classifier}    
    </protocArtifact>
    <pluginId>dubbo-grpc-java</pluginId>
    <pluginArtifact>org.apache.dubbo:protoc-gen-dubbo-java:1.19.0-SNAPSHOT:exe:${os.detected.classifier}</pluginArtifact>
    <outputDirectory>build/generated/source/proto/main/java</outputDirectory>
    <clearOutputDirectory>false</clearOutputDirectory>
    <pluginParameter>dubbo</pluginParameter>
  </configuration>
  <executions>
    <execution>
      <goals>
        <goal>compile</goal>
        <goal>compile-custom</goal>
      </goals>
    </execution>
  </executions>
</plugin>

1. 生成 RPC stub

# 运行以下 maven 命令
$mvn clean compile

public final class DemoServiceDubbo {
  private static final AtomicBoolean registered = new AtomicBoolean();
  private static Class<?> init() {
    Class<?> clazz = null;
    try {
      clazz = Class.forName(DemoServiceDubbo.class.getName());
      if (registered.compareAndSet(false, true)) {
        org.apache.dubbo.common.serialize.protobuf.support.ProtobufUtils.marshaller(
          org.apache.dubbo.demo.HelloRequest.getDefaultInstance());
        org.apache.dubbo.common.serialize.protobuf.support.ProtobufUtils.marshaller(
          org.apache.dubbo.demo.HelloReply.getDefaultInstance());
      }
    } catch (ClassNotFoundException e) {
      // ignore 
    }
    return clazz;
  }
  private DemoServiceDubbo() {}
  public static final String SERVICE_NAME = "demoservice.DemoService";
  /**
          * Code generated for Dubbo
          */
  public interface IDemoService {
    static Class<?> clazz = init();
    org.apache.dubbo.demo.HelloReply sayHello(org.apache.dubbo.demo.HelloRequest request);
    java.util.concurrent.CompletableFuture<org.apache.dubbo.demo.HelloReply> sayHelloAsync(
      org.apache.dubbo.demo.HelloRequest request);
  }
}

3. 开发业务逻辑

public class DemoServiceImpl implements DemoServiceDubbo.IDemoService {
  private static final Logger logger = LoggerFactory.getLogger(DemoServiceImpl.class);
  @Override
  public HelloReply sayHello(HelloRequest request) {
    logger.info("Hello " + request.getName() + ", request from consumer: " + RpcContext.getContext().getRemoteAddress());
    return HelloReply.newBuilder()
      .setMessage("Hello " + request.getName() + ", response from provider: "
                  + RpcContext.getContext().getLocalAddress())
      .build();
  }
  @Override
  public CompletableFuture<HelloReply> sayHelloAsync(HelloRequest request) {
    return CompletableFuture.completedFuture(sayHello(request));
  }
}

4. 配置 Provider

<dubbo:application name="demo-provider"/>
<dubbo:registry address="zookeeper://127.0.0.1:2181"/>
<dubbo:protocol name="dubbo"/>
<bean id="demoService" class="org.apache.dubbo.demo.provider.DemoServiceImpl"/>
<dubbo:service interface="org.apache.dubbo.demo.DemoServiceDubbo$IDemoService" ref="demoService"/>public static void main(String[] args) throws Exception {
  ClassPathXmlApplicationContext context = 
    new ClassPathXmlApplicationContext("spring/dubbo-provider.xml");
  context.start();
  System.in.read();
}

5. 配置 Consumer

<dubbo:application name="demo-consumer"/>
<dubbo:registry address="zookeeper://127.0.0.1:2181"/>
<dubbo:reference id="demoService" check="false" interface="org.apache.dubbo.demo.DemoServiceDubbo$IDemoService"/>public static void main(String[] args) throws Exception {
  ClassPathXmlApplicationContext context = 
    new ClassPathXmlApplicationContext("spring/dubbo-consumer.xml");
  context.start();
  IDemoService demoService = context.getBean("demoService", IDemoService.class);
  HelloRequest request = HelloRequest.newBuilder().setName("Hello").build();
  HelloReply reply = demoService.sayHello(request);
  System.out.println("result: " + reply.getMessage());
  System.in.read();
}

展望

来自 “ ITPUB博客 ” ，链接：http://blog.itpub.net/69947441/viewspace-2666498/，如需转载，请注明出处，否则将追究法律责任。