事件总线跨端桥接
你当前正在浏览历史版本 Vert.x 4.2.7 的文档。 访问本页面的 最新版本。

Vert.x Camel 网桥 (vertx-camel-bridge)

Apache Camel(http://camel.apache.org)是 Java 生态中久负盛名的开源企业集成模式(EIP)框架,致力于让异构系统间的互联集成更加优雅轻便。 Vert.x Camel 网桥建立起了 Vert.x 应用与 Camel 端点(Endpoint)之间的互通立交桥:

  • Vert.x 应用可以向 Camel 端点发送消息。

  • Vert.x 应用可以接收来自 Camel 端点的消息。

该网桥依托 Vert.x Event Bus,将特定的事件总线地址与 Camel 的 Endpoint 建立双向路由映射。

本组件不支持多语言跨端(Non-Polyglot),由于其深度依赖 Camel 原生 Java 类库,仅适用于 Java 开发环境。

引入依赖

要在项目中使用 Vert.x Camel 网桥,请在构建描述文件中添加以下依赖:

  • Maven(在 pom.xml 中):

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-camel-bridge</artifactId>
  <version>5.0.12</version>
</dependency>
  • Gradle(在 build.gradle 文件中):

compile 'io.vertx:vertx-camel-bridge:5.0.12'

网桥参数配置与启动

网桥在使用前必须完成映射配置并显式启动:

CamelContext camel = new DefaultCamelContext();
CamelBridge.create(vertx,
    new CamelBridgeOptions(camel)
        .addInboundMapping(InboundMapping.fromCamel("direct:stuff").toVertx("eventbus-address"))
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel("stream:out"))
).start();

网桥需要绑定一个 CamelContext 运行时上下文,并据此解析各类端点。请注意 start 方法是完全异步的,你可以利用返回的 Future 监听网桥启动就绪事件。

入站映射 (Inbound Mapping)

入站映射将一个 Camel 端点关联到一个 Event Bus 地址。在该 Camel 端点上接收到的消息将被自动转换为 Event Bus 消息并投递。

Endpoint endpoint = camel.getEndpoint("direct:foo");

CamelBridge.create(vertx,
    new CamelBridgeOptions(camel)
        .addInboundMapping(InboundMapping.fromCamel("direct:stuff").toVertx("eventbus-address"))
        .addInboundMapping(InboundMapping.fromCamel(endpoint).toVertx("eventbus-address"))
        .addInboundMapping(InboundMapping.fromCamel(endpoint).toVertx("eventbus-address")
            .withoutHeadersCopy())
        .addInboundMapping(InboundMapping.fromCamel(endpoint).toVertx("eventbus-address")
            .usePublish())
        .addInboundMapping(InboundMapping.fromCamel(endpoint).toVertx("eventbus-address")
            .withBodyType(String.class))
);

上述代码演示了入站映射的多种配置方式:

  • 既可以通过 Camel 端点 URI 字符串配置,也可以直接传入 Endpoint 对象

  • 可以禁用 Header 头信息的复制(默认会将 Camel Message Headers 自动复制到 Event Bus 消息中)

  • 可以配置使用 publish 广播替代默认的点对点 send,将消息分发给所有消费者

  • 可以显式配置 Event Bus 消息体的目标类型。若未指定,直接沿用 Camel 原始载荷;若指定,系统将在 Camel Context 中查找并调用对应的类型转换器(Converter)。

提示:org.fusesource.hawtbuf.Buffer 会被网桥自动透明转换为 Vert.x 原生的 Buffer。

如果使用 send(而非 publish),且 Camel Exchange 期望获得回包(In Out 请求-响应模式),Vert.x 代码将等待对发送消息的 Reply 应答。应答抵达后会自动传播回 Camel Exchange:

Endpoint endpoint = camel.getEndpoint("direct:stuff");

CamelBridge bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
    .addInboundMapping(new InboundMapping().setAddress("test-reply").setEndpoint(endpoint)));

vertx.eventBus().consumer("with-reply", message -> {
  message.reply("你最近怎么样?");
});

camel.start();
bridge.start();

ProducerTemplate template = camel.createProducerTemplate();
Future<Object> future = template.asyncRequestBody(endpoint, "hello");
String response = template.extractFutureBody(future, String.class);

你也可以通过 setTimeout 配置响应等待超时时间。

出站映射 (Outbound Mapping)

出站映射将一个 Event Bus 地址关联到一个 Camel 端点。发布或发送至该 Event Bus 地址的消息将被自动转换为 Camel 消息并转发至对应的 Camel 端点。

Endpoint endpoint = camel.getEndpoint("stream:out");

CamelBridge.create(vertx,
    new CamelBridgeOptions(camel)
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel("stream:out"))
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel(endpoint))
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel(endpoint)
            .withoutHeadersCopy())
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel(endpoint))
);

你可以将出站映射直接挂接至 Camel Route 路由:

camel.addRoutes(new RouteBuilder() {
  @Override
  public void configure() throws Exception {
    from("direct:start")
        .transform(constant("OK"));
  }
});

CamelBridge bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
    .addOutboundMapping(OutboundMapping.fromVertx("test").toCamel("direct:start")));

camel.start();
bridge.start();

vertx.eventBus().request("test", "hello").onComplete(reply -> {
  // 来自 Camel Route 的应答(此处为 "OK")
});

若在 Event Bus 发送消息时注册了 Reply Handler 应答回调,网桥会自动将 Camel Exchange 配置为请求-响应模式。响应内容将填入 Reply Body 中;若 Camel Route 处理失败,Event Bus 将收到失败回执:

camel.addRoutes(new RouteBuilder() {
  @Override
  public void configure() throws Exception {
    from("direct:my-route")
        .to("http://localhost:8080");
  }
});

CamelBridge bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
    .addOutboundMapping(OutboundMapping.fromVertx("camel-route").toCamel("direct:my-route")));

camel.start();
bridge.start();

vertx.eventBus().request("camel-route", "hello").onComplete(reply -> {
  if (reply.succeeded()) {
    Object theResponse = reply.result().body();
  } else {
    Throwable theCause = reply.cause();
  }
});

若 Camel 路由中包含耗时的阻塞处理,必须将 blocking 属性设为 true。这能确保阻塞逻辑派发至 Worker 线程执行,避免阻塞 Event Loop 响应式事件循环:

camel.addRoutes(new RouteBuilder() {
  @Override
  public void configure() throws Exception {
    from("direct:my-route")
      .process(new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
          // 执行耗时阻塞任务...
        }
      })
      .to("http://localhost:8080");
  }
});

CamelBridge bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
  .addOutboundMapping(OutboundMapping.fromVertx("camel-route").toCamel("direct:my-route").setBlocking(true)));

camel.start();
bridge.start();

vertx.eventBus().request("camel-route", "hello").onComplete(reply -> {
  if (reply.succeeded()) {
    Object theResponse = reply.result().body();
  } else {
    Throwable theCause = reply.cause();
  }
});

默认情况下阻塞任务提交给系统的通用 Worker 线程池,可以通过 setWorkerExecutor 定制指定的专用执行器。

停止网桥

在应用退出时,务必调用 stop 方法销毁网桥。stop 是异步方法,可通过返回的 Future 监听完全销毁事件。

传输自定义业务对象

若在 Event Bus 与 Camel 之间流转自定义 Java 对象,需在 Event Bus 上注册对应的编解码器(Codec):

vertx.eventBus().registerDefaultCodec(Person.class, codec);