服务工厂与辅助工具

你当前正在浏览历史版本 Vert.x 5.0.12 的文档。访问本页面的最新版本。

Vert.x gRPC

Deprecated

本项目已被 废弃（deprecated） 且后续将被移除，请改用 https://github.com/eclipse-vertx/vertx-grpc。

关于 gRPC 的最佳描述可见维基百科。

gRPC 是一个开源远程过程调用（RPC）系统，最初由 Google 开发。它使用 HTTP/2 作为传输层，使用 Protocol Buffers 作为接口描述语言，并提供身份认证、双向流式传输与流量控制、阻塞或非阻塞绑定以及请求取消与超时等特性。它可以为多种编程语言生成跨平台的客户端和服务端绑定。

— wikipedia
wikipedia

Vert.x gRPC 是一个将 Google gRPC 的编程风格与 Vert.x 风格对齐的模块。作为该模块的使用者，你可以在享受 gRPC 所有优势的同时，使用熟悉的 Vert.x Stream 和 Future 编码风格。

有关 gRPC 的更多信息，请查阅官方文档网站 http://www.grpc.io/。

此外，Vert.x gRPC 还支持：

通过 Verticle 实现 gRPC 服务的横向扩展
非阻塞原生传输层

gRPC 类型

借助 gRPC，你能够受益于 HTTP/2，这意味着你将获得异步流式传输支持，使得你的远程过程调用（RPC）可以具备以下特性：

客户端以流形式发送请求对象，服务端回复单个响应对象
客户端以流形式发送请求对象，服务端回复响应对象流
客户端发送单个请求对象，服务端回复单个响应对象
客户端发送单个请求对象，服务端回复响应对象流

乍看起来这与其他基于 HTTP 的 RPC 方案似乎差别不大，但你需要注意：在 HTTP/2 下，你的请求无需全部发送完毕即可开始接收响应。这意味着你的通信通道是全双工（full duplex）的。全双工能够有效降低响应延迟，并构建响应更及时的应用程序。

简单的 Hello World 示例

为了开始你的第一个 Hello World 示例，首先需要定义通信协议。gRPC 要求你使用 protobuffer 格式来定义该协议。

syntax = "proto3";

option java_multiple_files = true;
option java_package = "examples";
option java_outer_classname = "HelloWorldProto";
package helloworld;

// 问候服务定义
service Greeter {
  // 发送问候
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

// 包含用户名称的请求消息
message HelloRequest {
  string name = 1;
}

// 包含问候内容的响应消息
message HelloReply {
  string message = 1;
}

这是一个展示单请求、单响应模式的非常简单的示例。

编译 RPC 定义

完成上述定义后，我们需要对其进行编译。

如果你愿意，可以使用 protoc 编译器直接编译 proto 文件，或者也可以将其集成到你的构建工具中。

如果你使用的是 Apache Maven，需要添加以下插件：

 <plugin>
  <groupId>org.xolstice.maven.plugins</groupId>
  <artifactId>protobuf-maven-plugin</artifactId>
  <version>0.6.1</version>
  <configuration>
    <protocArtifact>com.google.protobuf:protoc:3.2.0:exe:${os.detected.classifier}</protocArtifact>
    <pluginId>grpc-java</pluginId>
    <pluginArtifact>io.grpc:protoc-gen-grpc-java:${vertx.grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
    <protocPlugins>
      <protocPlugin>
        <id>vertx-grpc-protoc-plugin</id>
        <groupId>io.vertx</groupId>
        <artifactId>vertx-grpc-protoc-plugin</artifactId>
        <version>${stack.version}</version>
        <mainClass>io.vertx.grpc.protoc.plugin.VertxGrpcGenerator</mainClass>
      </protocPlugin>
    </protocPlugins>
  </configuration>
  <executions>
    <execution>
      <id>compile</id>
      <configuration>
        <outputDirectory>${project.basedir}/src/main/java</outputDirectory>
        <clearOutputDirectory>false</clearOutputDirectory>
      </configuration>
      <goals>
        <goal>compile</goal>
        <goal>compile-custom</goal>
      </goals>
    </execution>
    <execution>
      <id>test-compile</id>
      <goals>
        <goal>test-compile</goal>
        <goal>test-compile-custom</goal>
      </goals>
    </execution>
  </executions>
</plugin>

${os.detected.classifier} 属性用于实现构建的操作系统无关性，在 OSX 上它会被替换为 osx-x86_64，依此类推。要使用该属性，你需要在 pom.xml 的 build 配置节中添加 os-maven-plugin[https://github.com/trustin/os-maven-plugin]：

<build>
  ...
  <extensions>
    <extension>
      <groupId>kr.motd.maven</groupId>
      <artifactId>os-maven-plugin</artifactId>
      <version>1.4.1.Final</version>
    </extension>
  </extensions>
  ...
</build>

该插件将编译 src/main/proto 目录下的 proto 文件，并使生成的代码在你的项目中可用。

如果你使用的是 Gradle，需要添加以下插件：

...
apply plugin: 'com.google.protobuf'
...
buildscript {
  ...
  dependencies {
    // 假设使用 Gradle 2.12 或更高版本。若使用较早版本的 Gradle，请改用 0.7.5 版本的插件
    classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.0'
  }
}
...
protobuf {
  protoc {
    artifact = 'com.google.protobuf:protoc:3.2.0'
  }
  plugins {
    grpc {
      artifact = "io.grpc:protoc-gen-grpc-java:1.25.0"
    }
    vertx {
      artifact = "io.vertx:vertx-grpc-protoc-plugin:${vertx.grpc.version}"
    }
  }
  generateProtoTasks {
    all()*.plugins {
      grpc
      vertx
    }
  }
}

该插件将编译 build/generated/source/proto/main 目录下的 proto 文件，并使生成的代码在你的项目中可用。

gRPC 服务端

现在你已经配置好了 RPC 基础代码，接下来开始实现服务端。正如前文所描述的，我们的服务端应当实现一个 sayHello 方法，该方法接收一个 HelloRequest 对象并返回一个 HelloReply 对象。你可以这样实现它：

GreeterGrpc.GreeterImplBase service = new GreeterGrpc.GreeterImplBase() {
  @Override
  public void sayHello(
    HelloRequest request,
    StreamObserver<HelloReply> responseObserver) {

    responseObserver.onNext(
      HelloReply.newBuilder()
        .setMessage(request.getName())
        .build());
    responseObserver.onCompleted();
  }
};

实现好服务后，你就可以将其发布到服务器上。Vert.x 使服务端的创建变得十分简单，你只需要添加以下代码：

VertxServer rpcServer = VertxServerBuilder
  .forAddress(vertx, "my.host", 8080)
  .addService(service)
  .build();

// 启动过程是异步的
rpcServer.start();

使用 Vert.x Future 与 Stream

前面的示例中使用的是基于 gRPC 原生异步结构（如 io.grpc.stub.StreamObserver）进行异步处理的 gRPC 服务端。这部分代码由 protoc 编译器生成。

上述插件配置中配置了如下插件：

<protocPlugin>
  <id>vertx-grpc-protoc-plugin</id>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-grpc-protoc-plugin</artifactId>
  <version>${stack.version}</version>
  <mainClass>io.vertx.grpc.protoc.plugin.VertxGrpcGenerator</mainClass>
</protocPlugin>

它会生成一个基于 Vert.x 异步结构（如 Future、ReadStream 或 WriteStream）的服务版本，这在 Vert.x 生态系统中会更加便捷顺手。

VertxGreeterGrpc.GreeterVertxImplBase service =
  new VertxGreeterGrpc.GreeterVertxImplBase() {
    @Override
    public Future<HelloReply> sayHello(HelloRequest request) {
      return Future.succeededFuture(
        HelloReply.newBuilder()
          .setMessage(request.getName())
          .build());
    }
  };

服务端 gzip 压缩

你可以启用 gzip 压缩，让服务端发送压缩后的响应（压缩请求由服务端自动处理）。

VertxGreeterGrpc.GreeterVertxImplBase service =
  new VertxGreeterGrpc.GreeterVertxImplBase() {
    @Override
    public Future<HelloReply> sayHello(HelloRequest request) {
      return Future.succeededFuture(
        HelloReply.newBuilder()
          .setMessage(request.getName())
          .build());
    }
  }
    .withCompression("gzip");

withCompression 配置由 Vert.x gRPC protoc 插件生成。你也可以在默认服务上启用压缩：只需将 ResponseObserver 强转为 ServerCallStreamObserver，并在发送响应之前调用 setCompression 即可。

GreeterGrpc.GreeterImplBase service = new GreeterGrpc.GreeterImplBase() {
  @Override
  public void sayHello(
    HelloRequest request,
    StreamObserver<HelloReply> responseObserver) {

    ((ServerCallStreamObserver) responseObserver)
      .setCompression("gzip");

    responseObserver.onNext(
      HelloReply.newBuilder()
        .setMessage(request.getName())
        .build());

    responseObserver.onCompleted();
  }
};

只要服务端支持，并且在构建 ManagedChannel 时已向压缩器注册表（compressor registry）完成注册，你就可以使用其他压缩算法。

SSL 配置

前面的示例很简单，但你的 RPC 尚未启用安全传输。为了保障安全性，我们应当启用 SSL/TLS：

VertxServerBuilder builder = VertxServerBuilder.forPort(vertx, 8080)
  .useSsl(options -> options
    .setSsl(true)
    .setUseAlpn(true)
    .setKeyCertOptions(new JksOptions()
      .setPath("server-keystore.jks")
      .setPassword("secret")));

恭喜，你刚刚完成了你的第一个 gRPC 服务端。

由于 gRPC 使用 HTTP/2 传输，因此服务端的 SSL/TLS 配置需要在你的服务端中启用应用层协议协商（ALPN）。

服务端横向扩展

当你部署同一 Verticle 的多个实例时，gRPC 服务端将在各 Verticle 的 Event Loop 上实现横向扩展。

vertx.deployVerticle(

  // Verticle 提供者 - 应被调用 4 次
  () -> new AbstractVerticle() {

    BindableService service = new GreeterGrpc.GreeterImplBase() {
      @Override
      public void sayHello(
        HelloRequest request,
        StreamObserver<HelloReply> responseObserver) {

        responseObserver.onNext(
          HelloReply.newBuilder()
            .setMessage(request.getName())
            .build());

        responseObserver.onCompleted();
      }
    };

    @Override
    public void start() throws Exception {
      VertxServerBuilder
        .forAddress(vertx, "my.host", 8080)
        .addService(service)
        .build()
        .start();
    }
  },

  // 部署 4 个实例，即服务将在 4 个 Event Loop 上横向扩展
  new DeploymentOptions()
    .setInstances(4));

BlockingServerInterceptor

gRPC ServerInterceptor 是一种在传入调用被分发到服务之前对其进行拦截的机制。它具有同步行为，并将在 Vert.x Event Loop 上执行。

VertxServer rpcServer = VertxServerBuilder
  .forAddress(vertx, "my.host", 8080)
  .addService(ServerInterceptors.intercept(service, myInterceptor))
  .build();

假设我们有一个会阻塞 Event Loop 的拦截器：

class MyInterceptor implements ServerInterceptor {
  @Override
  public <Q, A> ServerCall.Listener<Q> interceptCall(
    ServerCall<Q, A> call, Metadata headers, ServerCallHandler<Q, A> next) {
    // 执行耗时阻塞操作并更新元数据，例如：
    return next.startCall(call, headers);
  }
}
MyInterceptor myInterceptor = new MyInterceptor();

为了避免阻塞，应该对该拦截器进行包装。包装后它将在 Vert.x Worker 线程上被调用。

ServerInterceptor wrapped =
  BlockingServerInterceptor.wrap(vertx, myInterceptor);

// 创建服务端
VertxServer rpcServer = VertxServerBuilder
  .forAddress(vertx, "my.host", 8080)
  .addService(ServerInterceptors.intercept(service, wrapped))
  .build();

// 启动服务端
rpcServer.start();

Context Server Interceptor

这里提供了一个抽象的 Context 服务端拦截器，用于拦截服务端调用并将元数据提取到 Vert.x Context 中。该 Context 不依赖 ThreadLocal，因此可以安全地在 Vert.x API 中使用。该拦截器应当作为首个（或最先添加的几个之一）加入拦截器列表中。

一个典型的例子是会话 ID（session id）的使用。客户端可以创建一个 ClientInterceptor，为所有连接设置 session id：

Metadata extraHeaders = new Metadata();
extraHeaders.put(
  Metadata.Key.of("sessionId", Metadata.ASCII_STRING_MARSHALLER), theSessionId);

ClientInterceptor clientInterceptor = MetadataUtils
  .newAttachHeadersInterceptor(extraHeaders);

channel = VertxChannelBuilder.forAddress(vertx, "localhost", port)
  .intercept(clientInterceptor)
  .build();

接着在服务端，可以像下面这样添加拦截器：

BindableService service = new VertxGreeterGrpc.GreeterVertxImplBase() {
  @Override
  public Future<HelloReply> sayHello(HelloRequest request) {
    return Future.succeededFuture(
      HelloReply.newBuilder().setMessage("Hello " + request.getName()).build());
  }
};

ServerInterceptor contextInterceptor = new ContextServerInterceptor() {
  @Override
  public void bind(Metadata metadata) {
    put("sessionId", metadata.get(SESSION_ID_METADATA_KEY));
  }
};

// 创建服务端
VertxServer rpcServer = VertxServerBuilder
  .forAddress(vertx, "my.host", 8080)
  .addService(ServerInterceptors.intercept(service, contextInterceptor))
  .build();

gRPC 客户端

只有服务端而没有客户端是没有意义的，所以接下来我们创建一个客户端。为此，某些步骤与服务端会有重合。首先我们需要 RPC 定义文件，如果之前已经创建过服务端，那么这一步显然已经完成，且相同的定义也应该已经编译过了。

请注意，编译器总是会同时生成基础服务端与客户端存根（Stub），因此如果你之前已经编译过一次，就无需再次重新编译。

每个客户端存根始终需要一条通往服务端的通信通道，因此首先需要创建一个 gRPC Channel：

ManagedChannel channel = VertxChannelBuilder
  .forAddress(vertx, "localhost", 8080)
  .usePlaintext()
  .build();

// 获取用于与远程服务交互的 Stub
GreeterGrpc.GreeterStub stub = GreeterGrpc.newStub(channel);

存根创建完成后，我们就可以与服务端通信了。这一次更加轻松，因为存根已经提供了正确的方法定义和参数类型：

HelloRequest request = HelloRequest.newBuilder().setName("Julien").build();

// 调用远程服务
stub.sayHello(request, new StreamObserver<HelloReply>() {
  private HelloReply helloReply;

  @Override
  public void onNext(HelloReply helloReply) {
    this.helloReply = helloReply;
  }

  @Override
  public void onError(Throwable throwable) {
    System.out.println("Coult not reach server " + throwable.getMessage());
  }

  @Override
  public void onCompleted() {
    System.out.println("Got the server response: " + helloReply.getMessage());
  }
});

使用 Vert.x Future 与 Stream

前面的示例中使用的是基于 gRPC 原生异步结构（如 io.grpc.stub.StreamObserver）进行异步处理的 gRPC 客户端。这部分代码由 protoc 编译器生成。

上述插件配置中配置了如下插件：

<protocPlugin>
  <id>vertx-grpc-protoc-plugin</id>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-grpc-protoc-plugin</artifactId>
  <version>${stack.version}</version>
  <mainClass>io.vertx.grpc.protoc.plugin.VertxGrpcGenerator</mainClass>
</protocPlugin>

它会生成一个使用 Vert.x 异步结构（如 Future、ReadStream 或 WriteStream）的客户端版本，这在 Vert.x 生态系统中会更加便捷顺手。

HelloRequest request = HelloRequest.newBuilder().setName("Julien").build();

// 调用远程服务
Future<HelloReply> future = stub.sayHello(request);

// 监听完成事件
future
  .onSuccess(helloReply -> System.out.println("Got the server response: " + helloReply.getMessage())).onFailure(err -> System.out.println("Coult not reach server " + err));

客户端 gzip 压缩

你可以启用 gzip 压缩，让客户端发送经过压缩的消息。

GreeterGrpc.GreeterStub stub = GreeterGrpc
  .newStub(channel)
  .withCompression("gzip");

只要服务端支持，并且在构建 ManagedChannel 时已向压缩器注册表（compressor registry）完成注册，你就可以使用其他压缩算法。

SSL 配置

如果之前为服务端启用了 SSL，那么你的客户端也同样需要 SSL。为此我们需要配置 Channel：

ManagedChannel channel = VertxChannelBuilder.
  forAddress(vertx, "localhost", 8080)
  .useSsl(options -> options
    .setSsl(true)
    .setUseAlpn(true)
    .setKeyCertOptions(new JksOptions()
      .setPath("client-truststore.jks")
      .setPassword("secret")))
  .build();

由于 gRPC 使用 HTTP/2 传输，因此客户端的 SSL/TLS 配置需要在你的客户端中启用应用层协议协商（ALPN）。

高级配置

到目前为止，所有 gRPC 示例使用的都是合理的默认配置。但不仅限于此，如果你需要完全掌控服务端配置，可以参考文档：VertxServerBuilder；如果你需要控制客户端 Channel，可以参考 VertxChannelBuilder。Vert.x gRPC 扩展了 grpc-java 项目（Netty 传输层），因此推荐阅读其官方 Javadoc 文档。

原生传输层

客户端和服务端都可以基于 Netty 的原生传输层（Native Transports）进行部署，这在创建 Vert.x 实例时即可完成配置。

Vertx.vertx(new VertxOptions().setPreferNativeTransport(true));

有关原生传输层的更多信息，请参阅 Vert.x Core 文档。

gRPC

RxJava 3