事件总线 (Event Bus)

在 Event Bus 事件总线上交换传输 Protobuf 对象

本文档将向你展示如何在 Vert.x Event Bus 事件总线上直接收发由 Protocol Buffers (Protobuf) 生成的强类型消息。

你将构建什么

你将构建一个周期性生成问候消息的分布式应用程序，主要包含：

  • 一个负责发出 GreetingRequest 消息的发送端 Verticle；

  • 一个接收请求并以 GreetingResponse 进行回复的接收端 Verticle。

前置要求

  • 文本编辑器或 IDE

  • Java 11 或更高版本

创建工程项目

打开 https://start.vertx.io 并展开高级选项：

  • Group Id: 设置为 io.vertx.howtos

  • Artifact Id: 设置为 protobuf-eventbus-howto

  • Dependencies: 添加 Hazelcast Cluster Manager

  • Package: 设置为 io.vertx.howtos.protobuf.eventbus

或者通过命令行一键生成：

curl -G https://start.vertx.io/starter.zip -d "groupId=io.vertx.howtos" -d "artifactId=protobuf-eventbus-howto" -d "packageName=io.vertx.howtos.protobuf.eventbus" -d "vertxDependencies=vertx-hazelcast" -d "jdkVersion=11" -d "buildTool=maven" --output protobuf-eventbus-howto.zip
unzip -d protobuf-eventbus-howto protobuf-eventbus-howto.zip

定义 Protobuf 契约

在 src/main/proto/greetings.proto 中定义消息格式：

syntax = "proto3";

package greeting;

option java_multiple_files = true;
option java_package = "io.vertx.howtos.protobuf.eventbus";
option java_outer_classname = "GreetingProtos";

message GreetingRequest {
  string name = 1;
}

message GreetingReply {
  string message = 1;
}

实现消息编解码器

为了让 Event Bus 识别 Protobuf 生成的 Java 类，必须实现 Vert.x 的 MessageCodec 接口：

package io.vertx.howtos.protobuf.eventbus;

import com.google.protobuf.GeneratedMessageV3;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.MessageCodec;
import io.vertx.core.impl.SerializableUtils;

public class ProtobufCodec implements MessageCodec<GeneratedMessageV3, GeneratedMessageV3> {

  static final String PROTOS_PACKAGE_NAME = "io.vertx.howtos.protobuf.eventbus.";

  @Override
  public void encodeToWire(Buffer buffer, GeneratedMessageV3 o) {
    var bytes = SerializableUtils.toBytes(o);
    buffer.appendInt(bytes.length);
    buffer.appendBytes(bytes);
  }

  @Override
  public GeneratedMessageV3 decodeFromWire(int pos, Buffer buffer) {
    var length = buffer.getInt(pos);
    pos += 4;
    var bytes = buffer.getBytes(pos, pos + length);
    return (GeneratedMessageV3) SerializableUtils.fromBytes(bytes, CheckedClassNameObjectInputStream::new);
  }

  @Override
  public GeneratedMessageV3 transform(GeneratedMessageV3 o) {
    return o;
  }

  @Override
  public String name() {
    return "ProtobufCodec";
  }

  @Override
  public byte systemCodecID() {
    return -1; // -1 for a user codec
  }

  public boolean appliesTo(String className) {
    return className.startsWith(PROTOS_PACKAGE_NAME);
  }
}

在自定义 Launcher 中注册编解码器：

package io.vertx.howtos.protobuf.eventbus;

import io.vertx.launcher.application.HookContext;
import io.vertx.launcher.application.VertxApplication;
import io.vertx.launcher.application.VertxApplicationHooks;

public class CustomLauncher extends VertxApplication implements VertxApplicationHooks {

  public CustomLauncher(String[] args) {
    super(args);
  }

  public static void main(String[] args) {
    new CustomLauncher(args).launch();
  }

  @Override
  public void afterVertxStarted(HookContext context) {
    var vertx = context.vertx();
    var protobufCodec = new ProtobufCodec();
    vertx.eventBus().registerCodec(protobufCodec);
    vertx.eventBus().codecSelector(body -> {
      return protobufCodec.appliesTo(body.getClass().getName()) ? protobufCodec.name() : null;
    });
  }
}

实现发送者与接收者 Verticle

发送端 Verticle

package io.vertx.howtos.protobuf.eventbus;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.eventbus.Message;

public class SenderVerticle extends AbstractVerticle {

  @Override
  public void start() throws Exception {
    vertx.setPeriodic(5000, l -> {
      var request = GreetingRequest.newBuilder().setName("Jane Doe").build();
      System.out.printf("Sending request = %s (%d)%n", request.getName(), System.identityHashCode(request));
      vertx.eventBus().<GreetingReply>request("greetings", request)
        .map(Message::body)
        .onFailure(Throwable::printStackTrace)
        .onSuccess(reply -> System.out.printf("Received reply = %s (%d)%n", reply.getMessage(), System.identityHashCode(reply)));
    });
  }
}

接收端 Verticle

package io.vertx.howtos.protobuf.eventbus;

import io.vertx.core.Future;
import io.vertx.core.VerticleBase;

public class ReceiverVerticle extends VerticleBase {

  @Override
  public Future<?> start() {
    return vertx.eventBus().<GreetingRequest>consumer("greetings", msg -> {
      var request = msg.body();
      System.out.printf("Received request = %s (%d)%n", request.getName(), System.identityHashCode(request));
      var greeting = String.format("Hello %s", request.getName());
      var reply = GreetingReply.newBuilder().setMessage(greeting).build();
      System.out.printf("Sending reply = %s (%d)%n", reply.getMessage(), System.identityHashCode(reply));
      msg.reply(reply);
    }).completion();
  }
}

运行与验证

打开两个独立的终端窗口，分别启动接收端和发送端：

终端 1（接收端）：

./mvnw compile exec:java -Dexec.mainClass=io.vertx.howtos.protobuf.eventbus.CustomLauncher -Dexec.args="run io.vertx.howtos.protobuf.eventbus.ReceiverVerticle -cluster"

终端 2（发送端）：

./mvnw compile exec:java -Dexec.mainClass=io.vertx.howtos.protobuf.eventbus.CustomLauncher -Dexec.args="run io.vertx.howtos.protobuf.eventbus.SenderVerticle -cluster"

两个节点通过 Hazelcast 自动组网发现，你将在终端 2 看到接收端返回的清晰响应！

总结

本文档展示了：

  • 为 Google Protocol Buffers 编写通用的 Vert.x MessageCodec 消息编解码器；

  • 在集群环境下的 Event Bus 上安全高效传输二进制序列化对象。

延伸阅读