第三方框架整合
API 文档
你当前正在浏览历史版本 Vert.x 4.1.8 的文档。 访问本页面的 最新版本。

Vert.x STOMP 技术栈

STOMP 即简单（或流式）文本定向消息协议（Simple (or Streaming) Text Orientated Messaging Protocol）。 STOMP 提供了一种可互操作的网络连线格式，以便 STOMP 客户端能够与任何 STOMP 消息代理（Broker）进行通信， 从而在多种语言、平台以及消息代理之间实现简便且广泛的消息互操作性。有关 STOMP 的更多详细信息，请参阅 https://stomp.github.io/index.html 。

Vertx-Stomp 是 STOMP 服务端与客户端的 Vert.x 实现。你可以将此 STOMP 服务端与其他客户端搭配使用， 也可以将此 STOMP 客户端与其他服务端配合使用。该服务端和客户端支持 STOMP 协议的 1.0、1.1 以及 1.2 版本（请参见 https://stomp.github.io/stomp-specification-1.2.html ）。 此 STOMP 服务端还可以用作与 Vert.x EventBus 交互的网桥，或者直接配合 WebSocket（借助 StompJS）使用。

使用 vertx-stomp

若要使用 Vert.x STOMP 服务端和客户端，请将以下依赖项添加到项目构建描述文件的 dependencies 配置节中：

  • Maven（在 pom.xml 中）：

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-stomp</artifactId>
  <version>5.0.12</version>
</dependency>

  • Gradle（在 build.gradle 文件中）：

compile 'io.vertx:vertx-stomp:5.0.12'

STOMP 服务端

创建 STOMP 服务端

使用全部默认配置创建 STOMP 服务端的最简单方式如下：

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx))
    .listen();

这将创建一个监听在 localhost:61613 上并符合 STOMP 规范的 STOMP 服务端。

你可以在 listen 方法中配置端口与主机地址：

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx))
    .listen(1234, "0.0.0.0");

如果将端口传入 -1，则不会启动 TCP 服务端。这在使用 WebSocket 网桥时非常有用。 如果需要在服务端就绪时收到通知，可以使用如下 Handler：

StompServer server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx));
server
  .listen()
  .onComplete(ar -> {
      if (ar.failed()) {
        System.out.println("Failing to start the STOMP server : " + ar.cause().getMessage());
      } else {
        System.out.println("Ready to receive STOMP frames");
      }
    });

该 Handler 会接收一个对 StompServer 的引用。

你也可以在 StompServerOptions 中配置主机与端口：

Future<StompServer> server = StompServer.create(vertx, new StompServerOptions().setPort(1234).setHost("0.0.0.0"))
    .handler(StompServerHandler.create(vertx))
    .listen();

关闭 STOMP 服务端

关闭 STOMP 服务端的方式如下：

server
  .close()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("The STOMP server has been closed");
  } else {
    System.out.println("The STOMP server failed to close : " + ar.cause().getMessage());
  }
});

配置

StompServerOptions 允许你配置 STOMP 服务端的多个方面。

首先，STOMP 服务端基于 NetServer 构建，因此你可以通过 StompServerOptions 配置底层 NetServer。 此外，你也可以直接传入想要使用的 NetServer：

Future<StompServer> server = StompServer.create(vertx, netServer)
    .handler(StompServerHandler.create(vertx))
    .listen();

StompServerOptions 允许你配置：

  • STOMP 服务端的主机地址和端口 - 默认为 0.0.0.0:61613。

  • STOMP 服务端是否启用安全认证 - 默认为 false

  • STOMP Frame Body 的最大尺寸 - 默认为 10 Mb

  • STOMP Frame 中允许接收的最大 Header 数量 - 默认为 1000

  • STOMP Frame 中单行 Header 的最大长度 - 默认为 10240

  • STOMP 心跳时间 - 默认为 1000, 1000

  • 支持的 STOMP 协议版本（默认为 1.0、1.1 与 1.2）

  • 单个事务中允许的最大 Frame 数量（默认为 1000）

  • 事务分块大小（Chunk Size）- 默认为 1000（参见 setTransactionChunkSize）

  • 单个客户端可处理的最大 Subscription 数量 - 默认为 1000

STOMP 心跳使用 JSON 对象进行配置，如下所示：

Future<StompServer> server = StompServer.create(vertx, new StompServerOptions().setHeartbeat(
    new JsonObject().put("x", 1000).put("y", 1000)))
    .handler(StompServerHandler.create(vertx))
    .listen();

启用安全性需要额外的 AuthenticationProvider 来处理认证请求：

Future<StompServer> server = StompServer.create(vertx, new StompServerOptions().setSecured(true))
    .handler(StompServerHandler.create(vertx).authProvider(provider))
    .listen();

有关 AuthenticationProvider 的更多信息，请参阅 此处。

如果 Frame 超出任一大小限制，该 Frame 将被拒绝，客户端会收到一个 ERROR Frame。正如规范所要求的， 服务端在发送错误后会立即关闭客户端连接。超出其他阈值限制时也会发生相同的行为。

订阅

默认情况下，STOMP 服务端将订阅 Destination 视为不透明字符串（opaque String）。因此它不推行特定的结构， 也不具备层级结构。默认情况下，STOMP 服务端遵循 Topic（主题）语义（即消息会被分发给所有订阅者）。

Destination 类型

默认情况下，STOMP 服务端将 Destination 作为 Topic 管理。因此消息会被分发给所有订阅者。 你可以将服务端配置为使用队列（Queue），或者混合使用这两种类型：

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx)
        .destinationFactory((v, name) -> {
          if (name.startsWith("/queue")) {
            return Destination.queue(vertx, name);
          } else {
            return Destination.topic(vertx, name);
          }
        }))
    .listen();

在上一个示例中，所有以 /queue 开头的 Destination 均为队列，其他则为 Topic。当首次收到针对该 Destination 的 Subscription 时，Destination 会被创建。

服务端也可以通过返回 null 来拒绝创建 Destination：

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx)
        .destinationFactory((v, name) -> {
          if (name.startsWith("/forbidden")) {
            return null;
          } else if (name.startsWith("/queue")) {
            return Destination.queue(vertx, name);
          } else {
            return Destination.topic(vertx, name);
          }
        }))
    .listen();

在这种情况下，订阅者会收到一个 ERROR Frame。

队列使用轮询（round-robin）策略分发消息。

提供自定义 Destination 类型

STOMP 服务端特意未实现任何高级特性。如果你需要更高级的分发策略， 可以通过提供一个返回自定义 Destination 对象的 DestinationFactory 来实现自己的 Destination 类型。

消息确认（Acknowledgment）

默认情况下，当消息未被确认时，STOMP 服务端不做任何处理。你可以通过提供自定义的 Destination 实现来自定义该行为。

自定义 Destination 应调用

onAck 与 onNack 方法，以便让 StompServerHandler 自定义相关行为：

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx)
        .onAckHandler(acknowledgement -> {
          // 当 Frame（在 `client-individual` 模式下为 1 个，在 `client` 模式下为多个）被确认时执行的操作。
        })
        .onNackHandler(acknowledgement -> {
          // 当 Frame（在 `client-individual` 模式下为 1 个，在 `client` 模式下为多个）未被确认时执行的操作。
        }))
    .listen();

自定义 STOMP 服务端

除了上述 Handler 之外，你几乎可以配置 STOMP 服务端的方方面面，例如接收到特定 Frame 时执行的操作、发送给客户端的 ping（用于实现心跳）等。

以下是一些示例：

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx)
            .closeHandler(connection -> {
              // 客户端连接已关闭
            })
            .beginHandler(frame -> {
              // 事务开始
            })
            .commitHandler(frame -> {
                  // 事务已提交
                }
            )
        //...
    ).listen();

请注意，修改默认行为可能会破坏对 STOMP 规范的合规性。因此，请参考默认实现。

STOMP 客户端

STOMP 客户端用于连接到 STOMP 服务端，并能够发送和接收 Frame。

创建 STOMP 客户端

使用默认配置创建 StompClient 实例的方式如下：

StompClient.create(vertx)
  .connect()
  .onSuccess(connection -> {
    // 使用该连接
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

上述代码片段创建了一个连接到 "0.0.0.0:61613" 的 STOMP 客户端。连接成功后，你将获得一个 StompClientConnection，使你能够与服务端进行交互。你可以如下配置主机与端口：

StompClient.create(vertx)
  .connect(61613, "0.0.0.0")
  .onSuccess(connection -> {
    // 使用该连接
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

为了捕获因认证问题导致的连接错误，或服务端在连接协商期间发送的任何错误 Frame，你可以在 STOMP 客户端上注册一个 错误处理器（Error Handler）。使用该客户端创建的所有连接都会继承此错误处理器（也可以拥有各自独立的处理器）：

StompClient.create(vertx)
  .errorFrameHandler(frame -> {
    // 接收到 ERROR 帧
  })
  .connect(61613, "0.0.0.0")
  .onSuccess(connection -> {
    // 使用该连接
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

你也可以在 StompClientOptions 中配置主机与端口：

StompClient
  .create(vertx, new StompClientOptions().setHost("localhost").setPort(1234))
  .connect()
  .onSuccess(connection -> {
    // 使用该连接
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

关闭 STOMP 客户端

你可以关闭 STOMP 客户端：

StompClient client = StompClient
  .create(vertx, new StompClientOptions().setHost("localhost").setPort(1234));

client
  .connect()
  .onSuccess(connection -> {
    // 使用该连接
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

client.close();

但是，这种方式不会将断开连接的操作通知服务端。若要正常且干净地关闭连接，你应该使用 disconnect 方法：

StompClient
  .create(vertx, new StompClientOptions().setHost("localhost").setPort(1234))
  .connect()
  .onSuccess(connection -> {
    // 使用该连接
    connection.disconnect();
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

如果启用了心跳，并且客户端在配置的超时时间后未检测到服务端活动，连接将自动关闭。

错误处理

在 StompClientConnection 上，你可以注册一个错误处理器来接收服务端发送的 ERROR Frame。请注意，服务端在发送此类 Frame 后会关闭与客户端的连接：

StompClient
  .create(vertx, new StompClientOptions().setHost("localhost").setPort(1234))
  .connect()
  .onSuccess(connection -> {
    // 使用该连接
    connection
      .errorHandler(frame ->
        System.out.println("ERROR frame received : " + frame));
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

当检测到连接断开时，客户端也可以收到通知。连接故障是通过 STOMP 心跳机制检测的。当服务端在心跳时间窗口内未发送任何消息时，连接会被关闭，并调用 connectionDroppedHandler（如果已设置）。若要配置 connectionDroppedHandler，请调用 connectionDroppedHandler。例如，该 Handler 可以尝试重新连接到服务端：

StompClient.create(vertx)
  .connect()
  .onSuccess(connection -> {

    connection.connectionDroppedHandler(con -> {
      // 连接已丢失
      // 你可以重新连接或切换到其他服务端。
    });

    connection.send("/queue", Buffer.buffer("Hello"))
      .onSuccess(frame -> System.out.println("Message processed by the server")
      );
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

配置

你可以在创建 StompClient 时传入 StompClientOptions 来配置各项属性。由于 STOMP 客户端依赖于 NetClient，因此你可以通过 StompClientOptions 配置底层的 Net Client。

StompClientOptions 允许你配置：

  • STOMP 服务端的主机地址和端口

  • 连接服务端所需的 login 与 passcode

  • 当未显式设置时，是否自动为 Frame 添加 content-length Header（默认启用）

  • 是否使用 STOMP 命令代替 CONNECT 命令（默认禁用）

  • 是否在 CONNECT Frame 中忽略 host Header（默认禁用）

  • 心跳配置（默认为 1000, 1000）

订阅 Destination

若要订阅某个 Destination，请使用：

StompClient.create(vertx)
  .connect()
  .onSuccess(connection -> {
    // 使用该连接
    connection.subscribe("/queue", frame ->
      System.out.println("Just received a frame from /queue : " + frame));
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

若要取消订阅，请使用：

StompClient.create(vertx)
  .connect()
  .onSuccess(connection -> {
    // 使用该连接
    connection.subscribe("/queue", frame ->
      System.out.println("Just received a frame from /queue : " + frame));

    // ....

    connection.unsubscribe("/queue");
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

发送消息

若要发送消息，请使用：

StompClient.create(vertx)
  .connect()
  .onSuccess(connection -> {
    Map<String, String> headers = new HashMap<>();
    headers.put("header1", "value1");
    connection.send("/queue", headers, Buffer.buffer("Hello"));
    // 不带 Header：
    connection.send("/queue", Buffer.buffer("World"));
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

在 Java 和 Groovy 中，你可以使用 Headers 类来简化 Header 的创建。

确认机制（Acknowledgements）

客户端可以发送 ACK 和 NACK Frame：

StompClient.create(vertx)
  .connect()
  .onSuccess(connection -> {
    connection.subscribe("/queue", frame -> {
      connection.ack(frame.getAck());
      // 或者
      connection.nack(frame.getAck());
    });
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

事务

客户端还可以创建事务。在事务中发送的 ACK、NACK 和 SEND Frame 仅在事务提交时才会被交付。

StompClient.create(vertx)
  .connect()
  .onSuccess(connection -> {
    Map<String, String> headers = new HashMap<>();
    headers.put("transaction", "my-transaction");
    connection.beginTX("my-transaction");
    connection.send("/queue", headers, Buffer.buffer("Hello"));
    connection.send("/queue", headers, Buffer.buffer("World"));
    connection.send("/queue", headers, Buffer.buffer("!!!"));
    connection.commit("my-transaction");
    // 或者
    connection.abort("my-transaction");
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

收执（Receipt）

每个发送的命令都可以设置一个 收执（Receipt） 处理器，以便在服务端处理完该消息时收到通知：

StompClient.create(vertx)
  .connect()
  .onSuccess(connection -> {
    connection
      .send("/queue", Buffer.buffer("Hello"))
      .onSuccess(frame ->
        System.out.println("Message processed by the server"));
  })
  .onFailure(err ->
    System.out.println(
      "Failed to connect to the STOMP server: " + err.toString()));

将 STOMP 服务端用作 Vert.x EventBus 的网桥

STOMP 服务端可用作连接 Vert.x EventBus 的网桥。该网桥是双向的，这意味着 STOMP Frame 会被转换为 EventBus 消息，而 EventBus 消息也会被转换为 STOMP Frame。

要启用该网桥，你需要配置入站（inbound）和出站（outbound）地址。入站地址是指转发到 EventBus 的 STOMP Destination，该 STOMP Destination 将直接用作 EventBus 地址。出站地址是指转发到 STOMP 的 EventBus 地址。

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx)
        .bridge(new BridgeOptions()
            .addInboundPermitted(new PermittedOptions().setAddress("/toBus"))
            .addOutboundPermitted(new PermittedOptions().setAddress("/toStomp"))
        )
    )
    .listen();

默认情况下，网桥采用发布/订阅（Topic）交付模式。你也可以将其配置为点对点（Point to Point）交付模式，此时仅会调用一个 STOMP 客户端或 EventBus 消费者：

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx)
            .bridge(new BridgeOptions()
                    .addInboundPermitted(new PermittedOptions().setAddress("/toBus"))
                    .addOutboundPermitted(new PermittedOptions().setAddress("/toStomp"))
                    .setPointToPoint(true)
            )
    )
    .listen();

准入选项（Permitted Options）也可以表示为“正则表达式（regex）”或通过 match（匹配条件）指定。match 是消息载荷（Payload）必须满足的结构。例如在接下来的示例中，载荷必须包含值为 "bar" 的 "foo" 字段。结构匹配仅支持 JSON 对象。

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx)
        .bridge(new BridgeOptions()
            .addInboundPermitted(new PermittedOptions().setAddress("/toBus")
                .setMatch(new JsonObject().put("foo", "bar")))
            .addOutboundPermitted(new PermittedOptions().setAddress("/toStomp"))
            .setPointToPoint(true)
        )
    )
    .listen();

通过 WebSocket 使用 STOMP 服务端

如果你想将 JavaScript 客户端（Node.js 或浏览器环境）直接连接到 STOMP 服务端，可以使用 WebSocket。在 StompJS 中，STOMP 协议已被适配为可以通过 WebSocket 运行。JavaScript 可以直接连接到 STOMP 服务端，并在 WebSocket 上发送 STOMP Frame，同时也可以直接在 WebSocket 上接收 STOMP Frame。

要配置服务端使用 StompJS，你需要：

  1. 启用 WebSocket 网桥并配置监听 WebSocket 的路径（默认为 /stomp）。

  2. 在应用程序中引入 StompJS（作为 HTML 页面上的脚本，或作为 npm 模块：https://www.npmjs.com/package/stompjs ）。

  3. 连接到服务端。

要完成第一步，你需要一个 HTTP 服务端，并将 webSocketHandler 的返回结果传递给 webSocketHandler：

StompServer server = StompServer.create(vertx, new StompServerOptions()
    .setPort(-1) // 禁用 TCP 端口，可选
    .setWebsocketBridge(true) // 启用 WebSocket 支持
    .setWebsocketPath("/stomp")) // 配置 WebSocket 路径，默认为 /stomp
    .handler(StompServerHandler.create(vertx));

Future<HttpServer> http = vertx.createHttpServer(
    new HttpServerOptions().setWebSocketSubProtocols(Arrays.asList("v10.stomp", "v11.stomp"))
)
    .webSocketHandshakeHandler(server.webSocketHandshakeHandler())
    .webSocketHandler(server.webSocketHandler())
    .listen(8080);

切勿忘记声明支持的子协议（Sub-Protocols）。如果不声明，连接将被拒绝。

然后按照 StompJS 文档 中的说明连接到服务端。

配置 WebSocket Frame 类型

使用 WebSocket 网桥时，你可以配置 STOMP Frame 是作为 WebSocket 的 TEXT 帧还是 BINARY 帧发送。这对于与不同的 JavaScript STOMP 客户端保持兼容性非常重要：

StompServer server = StompServer.create(vertx, new StompServerOptions()
    .setPort(-1) // 禁用 TCP 端口，可选
    .setWebsocketBridge(true) // 启用 WebSocket 支持
    .setWebsocketPath("/stomp") // 配置 WebSocket 路径，默认为 /stomp
    .setWebSocketFrameType(WebSocketFrameType.TEXT)) // 为 JavaScript 客户端使用 TEXT 帧
  .handler(StompServerHandler.create(vertx));

Future<HttpServer> http = vertx.createHttpServer(
    new HttpServerOptions().setWebSocketSubProtocols(Arrays.asList("v10.stomp", "v11.stomp", "v12.stomp"))
  )
  .webSocketHandshakeHandler(server.webSocketHandshakeHandler())
  .webSocketHandler(server.webSocketHandler())
  .listen(8080);

WebSocketFrameType 枚举提供了两个选项：

  • TEXT - 将 STOMP Frame 作为 WebSocket 文本帧发送。推荐用于 JavaScript 客户端（例如 StompJS），因为 STOMP 是基于文本的协议。这是对 Web 浏览器兼容性最好的选项。

  • BINARY - 将 STOMP Frame 作为 WebSocket 二进制帧发送。这是较早版本中的默认行为，保留它是为了向后兼容。

当使用 TEXT 帧时，包含二进制数据的 STOMP 消息体必须是 UTF-8 安全的，或者应在应用层进行 Base64 编码。

注册接收与写入 Frame 处理器

STOMP 客户端、客户端连接以及服务端 Handler 都支持注册一个接收 Frame 的处理器，每当从网络连线（wire）接收到 Frame 时该处理器都会收到通知。这便于你记录 Frame 日志或实现自定义行为。该处理器针对 PING Frame 以及 非法/未知（illegal / unknown） Frame 同样会被调用：

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx).receivedFrameHandler(sf -> {
      System.out.println(sf.frame());
    }))
    .listen();

StompClient client = StompClient.create(vertx).receivedFrameHandler(frame -> System.out.println(frame));

该处理器会在 Frame 被处理之前被调用，因此你也可以借此 修改 Frame。

未使用有效 STOMP 命令的 Frame 将使用 UNKNOWN 命令。原始命令会通过 Frame.STOMP_FRAME_COMMAND 键记录在 Header 中。

你还可以注册一个处理器，以便在 Frame 即将被发送（写入到网络连线）时收到通知：

Future<StompServer> server = StompServer.create(vertx)
    .handler(StompServerHandler.create(vertx))
    .writingFrameHandler(sf -> {
      System.out.println(sf.frame());
    })
    .listen();

StompClient client = StompClient.create(vertx).writingFrameHandler(frame -> {
  System.out.println(frame);
});