消息队列与中间件
API 文档
你当前正在浏览历史版本 Vert.x 4.0.3 的文档。 访问本页面的 最新版本。

Vert.x MQTT

使用 Vert.x MQTT

该组件已正式发布在 Vert.x Stack 中，只需在构建描述文件的 dependencies 区域添加以下依赖即可：

  • Maven（在 pom.xml 中）：

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

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

compile "io.vertx:vertx-mqtt:5.0.12"

Vert.x MQTT 服务端

该组件提供了一个服务端，能够与远端 MQTT Client 处理连接、通信以及消息交换。其 API 提供了大量与 Client 接收到的原始协议消息相关的事件，并暴露了向 Client 发送消息的功能。

在撰写本文档时，它已支持 MQTT 5.0 版本的全部特性，仅除尚未实现的 AUTH 消息。

它不是一个功能完备的 MQTT Broker，但可用于构建类似系统或进行协议转换。

该模块处于技术预览（tech preview）阶段，这意味着不同版本之间的 API 可能会发生变更。

处理客户端连接与断开连接

本示例演示了如何处理来自远端 MQTT Client 的连接请求。首先创建一个 MqttServer 实例，并通过 endpointHandler 方法指定 Handler，当远端 Client 发送 CONNECT 消息以连接服务端时将调用该 Handler。作为参数传递给 Handler 的 MqttEndpoint 实例携带了与 CONNECT 消息相关的所有核心信息，例如 Client 标识符（client identifier）、用户名/密码、遗嘱（will）信息、清除会话（clean session）标志、协议版本、KeepAlive 超时时间以及 CONNECT 消息属性（针对 MQTT 5.0 版本）。 在该 Handler 内部，endpoint 实例提供了 accept 方法，用于向远端 Client 回复对应的 CONNACK 消息：连接就此建立。 最后，使用 listen 方法以默认行为启动服务端（监听 localhost 及默认 MQTT 端口 1883）。该方法同样允许指定一个 Handler，以便检查服务端是否成功启动。

MqttServer mqttServer = MqttServer.create(vertx);
mqttServer.endpointHandler(endpoint -> {

  // 显示主要连接信息
  System.out.println("MQTT client [" + endpoint.clientIdentifier() + "] request to connect, clean session = " + endpoint.isCleanSession());

  if (endpoint.auth() != null) {
    System.out.println("[username = " + endpoint.auth().getUsername() + ", password = " + endpoint.auth().getPassword() + "]");
  }
  System.out.println("[properties = " + endpoint.connectProperties() + "]");
  if (endpoint.will() != null) {
    System.out.println("[will topic = " + endpoint.will().getWillTopic() + " msg = " + new String(endpoint.will().getWillMessageBytes()) +
      " QoS = " + endpoint.will().getWillQos() + " isRetain = " + endpoint.will().isWillRetain() + "]");
  }

  System.out.println("[keep alive timeout = " + endpoint.keepAliveTimeSeconds() + "]");

  // 接受来自远端客户端的连接
  endpoint.accept(false);

})
  .listen()
  .onComplete(ar -> {

    if (ar.succeeded()) {

      System.out.println("MQTT server is listening on port " + ar.result().actualPort());
    } else {

      System.out.println("Error on starting the server");
      ar.cause().printStackTrace();
    }
  });

同一个 endpoint 实例提供了 disconnectMessageHandler 方法，用于指定当远端 Client 发送 DISCONNECT 消息以断开与服务端的连接时所调用的 Handler；该 Handler 接收 MqttDisconnectMessage 作为参数。

endpoint.disconnectMessageHandler(disconnectMessage -> {

  System.out.println("Received disconnect from client, reason code = " + disconnectMessage.code());
});

如果使用的是 MQTT 5.0 或更高版本，服务端可以通过 disconnect 方法向 Client 发送携带原因码和属性的 DISCONNECT 消息。

基于 SSL/TLS 支持处理客户端连接与断开连接

服务端支持通过 SSL/TLS 协议接收连接请求以实现认证与加密。为此，MqttServerOptions 类提供了 setSsl 方法用于开启 SSL/TLS（传入 true），并提供了其他实用方法用于配置服务端证书和相关的私钥（支持 Java KeyStore 引用、PEM 或 PFX 格式）。在以下示例中，使用 setKeyCertOptions 方法传入 PEM 格式的证书。该方法需要 KeyCertOptions 接口的实现实例，在此使用 PemKeyCertOptions 类，并通过相应的 setCertPath 与 setKeyPath 方法提供服务端证书和私钥的路径。创建 MQTT 服务端时，照常传入 Vert.x 实例以及上述 MQTT 配置选项实例即可。

MqttServerOptions options = new MqttServerOptions()
  .setPort(8883)
  .setKeyCertOptions(new PemKeyCertOptions()
    .setKeyPath("./src/test/resources/tls/server-key.pem")
    .setCertPath("./src/test/resources/tls/server-cert.pem"))
  .setSsl(true);

MqttServer mqttServer = MqttServer.create(vertx, options);
mqttServer.endpointHandler(endpoint -> {

  // 显示主要连接信息
  System.out.println("MQTT client [" + endpoint.clientIdentifier() + "] request to connect, clean session = " + endpoint.isCleanSession());

  if (endpoint.auth() != null) {
    System.out.println("[username = " + endpoint.auth().getUsername() + ", password = " + endpoint.auth().getPassword() + "]");
  }
  if (endpoint.will() != null) {
    System.out.println("[will topic = " + endpoint.will().getWillTopic() + " msg = " + new String(endpoint.will().getWillMessageBytes()) +
      " QoS = " + endpoint.will().getWillQos() + " isRetain = " + endpoint.will().isWillRetain() + "]");
  }

  System.out.println("[keep alive timeout = " + endpoint.keepAliveTimeSeconds() + "]");

  // 接受来自远端客户端的连接
  endpoint.accept(false);

})
  .listen()
  .onComplete(ar -> {

    if (ar.succeeded()) {

      System.out.println("MQTT server is listening on port " + ar.result().actualPort());
    } else {

      System.out.println("Error on starting the server");
      ar.cause().printStackTrace();
    }
  });

通过 WebSocket 处理客户端连接

如果你希望支持通过 WebSocket 进行连接，也可以通过 MqttServerOptions 来开启。通过向 setUseWebSocket 传入 true，服务端将在 /mqtt 路径上监听 WebSocket 连接。请求 URI 中可以包含查询字符串（Query String），例如 /mqtt?client=web，完整的 HTTP 请求 URI 可通过 httpRequestURI 获取。

与其他配置一样，由此产生的 Endpoint 连接及其断开处理与普通连接完全相同。

MqttServerOptions options = new MqttServerOptions()
  .setUseWebSocket(true);

MqttServer mqttServer = MqttServer.create(vertx, options);

处理客户端订阅与取消订阅请求

在 Client 与 Server 之间建立连接之后，Client 可以使用 SUBSCRIBE 消息发送针对某个 Topic 的订阅请求。MqttEndpoint 接口允许通过 subscribeHandler 方法为传入的订阅请求指定 Handler。该 Handler 接收 MqttSubscribeMessage 接口实例，其中包含了 Client 所期望的 Topic 列表以及相关的订阅选项。 订阅选项包括 QoS 等级和相关标志，针对 MQTT 5.0 版本还包括额外的标志，例如 noLocal 和 retainAsPublished。 最后，Endpoint 实例提供了 subscribeAcknowledge 方法，用于向 Client 回复相关的 SUBACK 消息，其中包含原因码（每个 Topic 或匹配模式对应的 QoS 等级或错误码）和消息属性。

endpoint.subscribeHandler(subscribe -> {

  List<MqttSubAckReasonCode> reasonCodes = new ArrayList<>();
  for (MqttTopicSubscription s: subscribe.topicSubscriptions()) {
    System.out.println("Subscription for " + s.topicName() + " with QoS " + s.qualityOfService());
    reasonCodes.add(MqttSubAckReasonCode.qosGranted(s.qualityOfService()));
  }
  // 确认订阅请求
  endpoint.subscribeAcknowledge(subscribe.messageId(), reasonCodes, MqttProperties.NO_PROPERTIES);

});

同样地，可以在 Endpoint 上使用 unsubscribeHandler 方法指定当 Client 发送 UNSUBSCRIBE 消息时调用的 Handler。该 Handler 接收 MqttUnsubscribeMessage 接口实例作为参数，其中包含待取消订阅的 Topic 列表。 最后，Endpoint 实例提供了 unsubscribeAcknowledge 和 unsubscribeAcknowledge 方法，用于向 Client 回复相关的 UNSUBACK 消息 —— 既可以简单地对所有取消订阅进行确认，也可以针对每个 Topic 指定原因码以及 UNSUBSCRIBE 请求中的属性（在 MQTT 5.0 及更高版本中受支持）。

endpoint.unsubscribeHandler(unsubscribe -> {

  for (String t: unsubscribe.topics()) {
    System.out.println("Unsubscription for " + t);
  }
  // 确认取消订阅请求
  endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
});

处理客户端发布的消息

为了处理远端 Client 发布的传入消息，MqttEndpoint 接口提供了 publishHandler 方法，用于指定当 Client 发送 PUBLISH 消息时调用的 Handler。该 Handler 接收 MqttPublishMessage 接口实例作为参数，其中包含 Payload、QoS 等级、重发（duplicate）与保留（retain）标志以及消息属性。

如果 QoS 等级为 0（AT_MOST_ONCE），Endpoint 无需向 Client 进行回复。

如果 QoS 等级为 1（AT_LEAST_ONCE），Endpoint 需要使用可用的 publishAcknowledge 或 publishAcknowledge 方法回复 PUBACK 消息。

如果 QoS 等级为 2（EXACTLY_ONCE），Endpoint 需要使用可用的 publishReceived 或 publishReceived 方法回复 PUBREC 消息；在这种情况下，同一个 Endpoint 还需要处理从 Client 接收到的 PUBREL 消息（远端 Client 在收到 Endpoint 发送的 PUBREC 之后会发送该消息），这可以通过 publishReleaseHandler 或 publishReleaseMessageHandler 方法指定 Handler —— 具体取决于服务端是否需要访问 MQTT 5.0 的扩展能力（原因码、消息属性）。 为了完成 QoS 2 等级的投递闭环，Endpoint 可以使用 publishComplete 或 publishComplete 方法向 Client 发送 PUBCOMP 消息。

endpoint.publishHandler(message -> {

  System.out.println("Just received message [" + message.payload().toString(Charset.defaultCharset()) + "] with QoS [" + message.qosLevel() + "]");

  if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
    endpoint.publishAcknowledge(message.messageId());
  } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) {
    endpoint.publishReceived(message.messageId());
  }

}).publishReleaseHandler(messageId -> {

  endpoint.publishComplete(messageId);
});

向客户端发布消息

Endpoint 可以使用 publish 方法向远端 Client 发布消息（发送 PUBLISH 消息），该方法接收以下输入参数：待发布的 Topic、Payload、QoS 等级、重发标志和保留标志。 如果你正在使用 MQTT 5.0 或更高版本并且想要指定消息属性，可以使用 publish 方法，该方法在前述方法的基础上额外接收消息 ID 和消息属性。

如果 QoS 等级为 0（AT_MOST_ONCE），Endpoint 不会收到来自 Client 的任何反馈。

如果 QoS 等级为 1（AT_LEAST_ONCE），Endpoint 需要处理从 Client 接收到的 PUBACK 消息，以获得最终的投递确认。这可以通过 publishAcknowledgeHandler 或 publishAcknowledgeMessageHandler 方法指定此类 Handler 来实现。

如果 QoS 等级为 2（EXACTLY_ONCE），Endpoint 需要处理从 Client 接收到的 PUBREC 消息。publishReceivedHandler 与 publishReceivedMessageHandler 方法允许为此指定 Handler。在该 Handler 内部，Endpoint 可以使用 publishRelease 或 publishRelease 方法向 Client 回复 PUBREL 消息。最后一步是处理从 Client 接收到的 PUBCOMP 消息作为已发布消息的最终确认；这可以通过 publishCompletionHandler 或 publishCompletionMessageHandler 方法指定当收到最终的 PUBCOMP 消息时所调用的 Handler。

endpoint.publish("my_topic",
  Buffer.buffer("Hello from the Vert.x MQTT server"),
  MqttQoS.EXACTLY_ONCE,
  false,
  false);

// 指定用于处理 QoS 1 和 2 的 Handler
endpoint.publishAcknowledgeHandler(messageId -> {

  System.out.println("Received ack for message = " +  messageId);

}).publishReceivedHandler(messageId -> {

  endpoint.publishRelease(messageId);

}).publishCompletionHandler(messageId -> {

  System.out.println("Received ack for message = " +  messageId);
});

接收客户端 KeepAlive 保活通知

底层的 MQTT KeepAlive 保活机制由服务端在内部处理。当收到 CONNECT 消息时，服务端会关注该消息中指定的 KeepAlive 超时时间，以检查 Client 是否在该超时期限内未发送消息。与此同时，每当收到 PINGREQ 时，服务端都会回复相应的 PINGRESP。

即便高层应用无需自行处理该逻辑，MqttEndpoint 接口依然提供了 pingHandler 方法，用于指定当从 Client 收到 PINGREQ 消息时调用的 Handler。这仅用于向应用程序发出通知，表明 Client 当前并未发送有意义的业务消息，而只是通过 Ping 来保持连接；无论如何，如前所述，PINGRESP 都会由服务端在内部自动回复。

endpoint.pingHandler(v -> {

  System.out.println("Ping received from client");
});

关闭服务端

MqttServer 接口提供了 close 方法用于关闭服务端；它会停止监听传入的连接并关闭与远端 Client 的所有活动连接。该方法是异步的，并且提供了一个重载版本允许指定完成 Handler，在服务端完全关闭后该 Handler 将被调用。

mqttServer.close().onComplete(v -> {

  System.out.println("MQTT server closed");
});

处理客户端认证报文 / 向远端客户端发送 AUTH 报文（仅限 MQTT 5.0）

在 Client 与 Server 之间建立连接之后，Client 可以使用 AUTH 消息向 Server 发送认证报文。MqttEndpoint 接口允许通过 authenticationExchangeHandler 方法为传入的认证报文指定 Handler。该 Handler 接收 MqttAuthenticationExchangeMessage 接口实例，其中携带原因码、认证方法和数据。服务端可以使用 authenticationExchange 继续发送 AUTH 报文以进行认证，或者直接放行。

endpoint.authenticationExchange(MqttAuthenticationExchangeMessage.create(MqttAuthenticateReasonCode.SUCCESS, MqttProperties.NO_PROPERTIES));
// 处理来自客户端的认证信息
endpoint.authenticationExchangeHandler(auth -> {
  System.out.println("AUTH packet received from client. code: " + auth.reasonCode());
});

在 Verticle 中自动清理

如果你是在 Verticle 内部创建 MQTT 服务端，当该 Verticle 被卸载（undeploy）时，这些服务端将被自动关闭。

伸缩：共享 MQTT 服务端

与 MQTT 服务端相关的 Handler 始终在同一个 Event Loop 线程中执行。这意味着在多核系统上，如果仅部署了一个实例，就只会利用一个核心。为了利用更多核心，可以部署多个 MQTT 服务端实例。

可以通过编程方式实现：

for (int i = 0; i < 10; i++) {

  MqttServer mqttServer = MqttServer.create(vertx);
  mqttServer.endpointHandler(endpoint -> {
    // 处理 endpoint
  })
    .listen()
    .onComplete(ar -> {

      // 处理开始监听的结果
    });

}

或者使用 Verticle 并指定实例数量：

DeploymentOptions options = new DeploymentOptions().setInstances(10);
vertx.deployVerticle("com.mycompany.MyVerticle", options);

实际的工作机制是：即便仅部署了单个逻辑 MQTT 服务端，但当新连接到来时，Vert.x 会以轮询（round-robin）的方式将它们分发到运行在不同核心上的各个连接 Handler 中。

Vert.x MQTT 客户端

该组件提供了兼容 3.1.1 与 5.0 规范的 MQTT Client。其 API 提供了大量用于连接/断开 Broker、发布消息（支持全部三种不同的 QoS 等级）以及订阅 Topic 的方法。在 3.1.1 基线之上，还支持 MQTT 5.0 的用户属性（User Properties）、原因码（Reason Codes）、订阅选项（Subscription Options）、订阅标识符（Subscription Identifiers）、Topic 别名（Topic Aliases）以及自动服务端重定向等特性。

该模块处于技术预览（tech preview）阶段，这意味着不同版本之间的 API 可能会发生变更。

连接与断开连接

Client 为你提供了连接到服务端以及从服务端断开连接的能力。此外，你还可以通过向构造函数传入 MqttClientOptions 实例作为参数，来指定要连接的服务端主机名和端口等信息。

以下示例演示了如何使用 Vert.x MQTT Client 并调用 connect 与 disconnect 方法来连接和断开服务端。

MqttClient client = MqttClient.create(vertx);

client.connect(1883, "mqtt.eclipse.org").onComplete(s -> {
  client.disconnect();
});
由 MqttClientOptions 提供的服务端默认地址为 localhost:1883，如果你使用 SSL/TLS，则为 localhost:8883。

订阅 Topic

现在，让我们进一步深入并看一看这个示例：

client.publishHandler(s -> {
  System.out.println("There are new message in topic: " + s.topicName());
  System.out.println("Content(as string) of the message: " + s.payload().toString());
  System.out.println("QoS: " + s.qosLevel());
})
  .subscribe("rpi2/temp", 2);

这里展示了 subscribe 方法的使用示例。为了接收来自 rpi2/temp Topic 的消息，我们调用了 subscribe 方法。 此外，要处理从服务端接收到的消息，你需要提供一个 Handler，每当所订阅的 Topic 中出现新消息时，该 Handler 就会被调用。 如本例所示，Handler 可以通过 publishHandler 方法提供。

向 Topic 发布消息

如果你想向 Topic 发布消息，应当调用 publish。 让我们看一个示例：

client.publish("temperature",
  Buffer.buffer("hello"),
  MqttQoS.AT_LEAST_ONCE,
  false,
  false);

在该示例中，我们向名为 "temperature" 的 Topic 发送了一条消息。

处理服务端认证请求 / 向服务端发送 AUTH 报文（仅限 MQTT 5.0）

在 Client 与 Server 之间建立连接之后，Client 可以使用 authenticationExchange 向 Server 发送认证请求。Server 可能会返回 AUTH 报文。MqttClient 接口允许通过 authenticationExchangeHandler 方法为传入的认证报文指定 Handler。该 Handler 接收 MqttAuthenticationExchangeMessage 接口实例，其中携带原因码、认证方法和数据。

client.authenticationExchange(MqttAuthenticateReasonCode.SUCCESS, MqttProperties.NO_PROPERTIES);
client.authenticationExchangeHandler(auth -> {
  // 默认情况下该 Handler 会被不定期调用
  System.out.println("We have just received AUTH packet: " + auth.reasonCode());
});

保持与服务端的连接存活

为了保持与服务端的连接，你需要不定期向服务端发送数据，否则服务端会关闭该连接。 保持连接存活的正确方式是使用 ping 方法。

默认情况下，Client 会自动保持与服务端的连接。这意味着你无需为了保持与服务端的连接而主动调用 ping，MqttClient 会自动为你完成。

如果你想禁用此特性，可以将 false 作为参数调用 setAutoKeepAlive：

options.setAutoKeepAlive(false);

接收通知的时机

  • 发布完成

    你可以通过调用 publishCompletionHandler 提供 Handler。每次发布完成时该 Handler 都会被调用。 这非常实用，因为你可以查看到刚收到的 PUBACK 或 PUBCOMP 报文的 packetId。

    client.publishCompletionHandler(id -> {
  System.out.println("Id of just received PUBACK or PUBCOMP packet is " + id);
});
  // 下面这行代码会触发 publishCompletionHandler（QoS 2）
client.publish("hello", Buffer.buffer("hello"), MqttQoS.EXACTLY_ONCE, false, false);
  // 下面这行代码会触发 publishCompletionHandler（QoS 为 1）
client.publish("hello", Buffer.buffer("hello"), MqttQoS.AT_LEAST_ONCE, false, false);
  // 下面这行代码不会触发，因为 QoS 值为 0
client.publish("hello", Buffer.buffer("hello"), MqttQoS.AT_LEAST_ONCE, false, false);
    如果发送的发布报文 QoS=0，该 Handler 将不会被调用。

  • 订阅完成

    client.subscribeCompletionHandler(mqttSubAckMessage -> {
  System.out.println("Id of just received SUBACK packet is " + mqttSubAckMessage.messageId());
  for (int s : mqttSubAckMessage.grantedQoSLevels()) {
    if (s == 0x80) {
      System.out.println("Failure");
    } else {
      System.out.println("Success. Maximum QoS is " + s);
    }
  }
});
client.subscribe("temp", 1);
client.subscribe("temp2", 2);

  • 取消订阅完成

    client
  .unsubscribeCompletionHandler(id -> {
    System.out.println("Id of just received UNSUBACK packet is " + id);
  });
client.subscribe("temp", 1);
client.unsubscribe("temp");

  • 取消订阅已发送

    client.subscribe("temp", 1);
client.unsubscribe("temp").onSuccess(id -> {
    System.out.println("Id of just sent UNSUBSCRIBE packet is " + id);
  });

  • 接收到 PINGRESP

    client.pingResponseHandler(s -> {
  // 默认情况下该 Handler 会被不定期调用
  System.out.println("We have just received PINGRESP packet");
});

使用 TLS 进行连接

你可以通过配置 Client 的 TCP 选项来使用 TLS 连接到 MQTT 服务端，请确保完成以下设置：

  • 开启 SSL 标志

  • 配置服务端证书或开启信任所有证书（trust all）标志

  • 如果需要校验服务端身份，将主机名验证算法设置为 "HTTPS"，否则设置为 ""

MqttClientOptions options = new MqttClientOptions();
options
  .setSsl(true)
  .setTrustOptions(new PemTrustOptions().addCertPath("/path/to/server.crt"))
  // 算法可以为空字符串 ""，或者为 "HTTPS" 以验证服务端主机名
  .setHostnameVerificationAlgorithm(algo);
MqttClient client = MqttClient.create(vertx, options);
client.connect(1883, "mqtt.eclipse.org").onComplete(s -> {
  client.disconnect();
});
关于 TLS 客户端配置的更多详细信息，请参阅 此处

使用代理协议（Proxy Protocol）

MqttServer mqttServer = MqttServer
  .create(vertx, new MqttServerOptions()
    // 设置为 true 以使用代理协议
    .setUseProxyProtocol(true));
mqttServer.endpointHandler(endpoint -> {
  // 远端地址为原始真实地址，而非代理的地址
  System.out.println(endpoint.remoteAddress());
  endpoint.accept(false);

})
  .listen()
  .onComplete(ar -> {

    if (ar.succeeded()) {

      System.out.println("MQTT server is listening on port " + ar.result().actualPort());
    } else {

      System.out.println("Error on starting the server");
      ar.cause().printStackTrace();
    }
  });

如果你的服务端部署在 HAProxy 或 Nginx 等代理之后，并且你希望获取 Client 的原始 IP 与端口，那么需要将 setUseProxyProtocol 设置为 true。

要启用此特性，你需要添加依赖 netty-codec-haproxy，但该依赖并未默认引入，因此需要手动添加：

  • Maven（在 pom.xml 中）：

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-codec-haproxy</artifactId>
    <version>5.0.12</version>
</dependency>

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

compile "io.netty:netty-codec-haproxy:5.0.12"

MQTT 5.0 客户端特性

本章节为全新内容，目前处于技术预览（tech preview）阶段，换句话说，在达到成熟状态之前，API 可能会发生变动。

Client 默认以 MQTT 3.1.1 为目标协议。若要将传输协议切换为 MQTT 5.0，请在创建 Client 之前在 MqttClientOptions 上设置协议版本：

MqttClientOptions options = new MqttClientOptions();
options.setVersion(io.netty.handler.codec.mqtt.MqttVersion.MQTT_5.protocolLevel()); // 5
MqttClient client = MqttClient.create(vertx, options);

setVersion 仅接受 4（MQTT 3.1.1）与 5（MQTT 5.0）两个值。

CONNECT 属性

使用 MQTT 5.0 时，可以在 MqttClientOptions 上配置以下 CONNECT 属性：

可以通过接收用户属性 Map<String, String> 的 connect 重载方法，将用户属性（User Properties）附加到 CONNECT 报文中。

CONNECT 的结果以 MqttConnAckMessage 的形式交付，在 MQTT 5.0 中，该对象公开了服务端通告的全套属性（最大接收量 Receive Maximum、最大 QoS Maximum QoS、保留可用性 Retain Available、最大报文大小 Maximum Packet Size、分配的客户端标识符 Assigned Client Identifier、最大 Topic 别名 Topic Alias Maximum、原因字符串 Reason String、通配符/共享订阅可用性 Wildcard / Shared Subscription Available、订阅标识符可用性 Subscription Identifiers Available、服务端 KeepAlive Server Keep Alive、响应信息 Response Information、服务端引用 Server Reference、认证方法/数据 Authentication Method/Data 以及用户属性）。

会话过期时间（Session Expiry Interval）

MQTT 5.0 的会话过期时间（单位为秒）取代了 3.1.1 的 清除会话（clean session） 标志，它告知 Broker 在网络连接关闭之后会话状态（订阅、排队的 QoS 1/2 消息）应当保留多长时间。连接前在 MqttClientOptions 上进行设置：

MqttClientOptions options = new MqttClientOptions();
options.setVersion(io.netty.handler.codec.mqtt.MqttVersion.MQTT_5.protocolLevel());
options.setCleanSession(false);                  // 保持持久会话所需
options.setSessionExpireInterval(3600L);         // 断开连接后将该会话保留 1 小时
MqttClient client = MqttClient.create(vertx, options);

其语义遵循 MQTT 5.0 §3.1.2.11：

  • null（默认值）或 0L —— 网络连接一旦断开，会话立即结束（等价于 clean session）。

  • 正数值 —— 连接关闭后，会话将继续保留该秒数。

  • 0xFFFFFFFFL（4_294_967_295L）—— 会话永不过期。

有效取值范围为 0L..0xFFFFFFFFL；超出此范围的值会导致 setSessionExpireInterval 抛出 IllegalArgumentException。该属性仅在协议版本为 5 时才会写入网络报文；在 MQTT 3.1.1 中该值会被静默忽略，会话持久化完全由 setCleanSession 驱动。

Broker 可能会覆盖所请求的值：当 CONNACK 中返回 Session Expiry Interval 时，该值在会话生命周期内具有最终权威性，可通过 sessionExpiryInterval 进行查看。

最大接收量（Receive Maximum）

最大接收量（Receive Maximum，MQTT 5.0 §3.1.2.11.3）是针对 QoS 1 和 QoS 2 PUBLISH 报文的单向流控窗口：它是对端愿意接收的进行中（in-flight，已发送但尚未完全确认）消息的最大数量。通信双方各自通告自身的上限，且必须遵守对端通告的上限。

客户端 —— 连接前在 MqttClientOptions 上设置：

MqttClientOptions options = new MqttClientOptions();
options.setVersion(io.netty.handler.codec.mqtt.MqttVersion.MQTT_5.protocolLevel());
options.setReceiveMaximum(20); // 最多同时接收来自服务端的 20 条进行中 QoS 1/2 PUBLISH 消息
MqttClient client = MqttClient.create(vertx, options);

  • 该值为 16 位正整数；有效范围为 1..65535。超出 0..0xFFFF 的值会被 setReceiveMaximum 拒绝并抛出 IllegalArgumentException。

  • null（默认值）会在 CONNECT 中省略该属性，根据规范这代表“无限制”（65535）。

  • 该属性仅在协议版本为 5 时才会写入网络报文；在 MQTT 3.1.1 中该值会被忽略。

服务端 —— Broker 在 CONNACK 中通告的值在为 QoS 1 或 2 调用 publish 时会自动强制生效：

  • 如果进行中的消息数量已达到服务端的 Receive Maximum，返回的 Future 将失败，并附带错误码为 MQTT_INFLIGHT_QUEUE_FULL 的 MqttException；

  • QoS 0 PUBLISH 报文不受此限制门控（它们无需确认）；

  • 如果 CONNACK 中省略了该属性，Client 将其视为无限制（内部为 Integer.MAX_VALUE）；

  • 协商后的数值可以通过 receiveMaximum 查看。

该限制独立于通过 setMaxInflightQueue 配置的客户端本地现有上限（并在此之上生效）；无论哪一个限制先被触发，都会抛出 MQTT_INFLIGHT_QUEUE_FULL 失败。

遗嘱消息（Last Will and Testament, LWT）

遗嘱消息配置已被移至专门的 MqttClientWillOptions 对象中，该对象同时支持 3.1.1 字段（Topic、Payload、QoS、保留标志）以及 MQTT 5.0 的遗嘱属性：遗嘱延迟间隔（Will Delay Interval）、Payload 格式指示符（Payload Format Indicator）、内容类型（Content Type）、响应主题（Response Topic）、相关数据（Correlation Data）和用户属性。遗嘱选项可以通过 setWillOptions 进行设置；旧版的 setWillTopic / setWillMessage / setWillQoS / setWillRetain Setter 依然可用以保持与 3.1.1 的兼容性。

Client 端的独立 willFlag Setter 已被移除 —— 是否存在遗嘱现在由 willTopic 和 willPayload 共同推导。在 JSON 中，遗嘱被序列化为嵌套的 willOptions 对象。

携带属性进行发布

使用 MQTT 5.0 时，发出的 PUBLISH 报文可以携带额外属性（Payload 格式指示符、消息过期时间、内容类型、响应主题、相关数据、用户属性、Topic 别名）。可以使用 publish 在 Payload 的同时传入 MqttProperties 实例。订阅标识符（Subscription Identifier）属性在 Client 发起的 PUBLISH 上是 无效 的 —— 请参阅下方的专门章节。

确认流现在暴露了包含原因码和属性的完整强类型消息：

这些 Handler 会与现有的 publishCompletionHandler 一同触发，从而保证 3.1.1 代码能够继续正常工作。 在入站侧，Client 还可以通过以下方法携带原因码和属性对传入的 PUBLISH 进行确认：publishAcknowledge、publishReceived、publishRelease 以及 publishComplete。

订阅（Subscriptions）

提供了两个 MQTT 5.0 特有的 subscribe 重载版本：

  • subscribe —— 具有与 3.1.1 API 相同的 QoS 映射，并外加一个属性对象（例如 Subscription Identifier）。

  • subscribe —— 用于细粒度的订阅选项。每个 MqttTopicSubscription 都携带一个 MqttSubscriptionOption，它编码了 QoS 以及 v5 标志（No Local、Retain As Published、Retain Handling）。

unsubscribe 具有对应的接收属性的 unsubscribe 重载。SUBACK 和 UNSUBACK 响应通过 MqttSubAckMessage 和 MqttUnsubAckMessage 暴露每个 Topic 的原因码和属性；对于 UNSUBACK，专门的 unsubscribeCompletionMessageHandler 会交付完整的强类型消息。

订阅标识符（Subscription Identifier）

订阅标识符（Subscription Identifier，MQTT 5.0 §3.8.2.1.2）是 Client 附加到 SUBSCRIBE 请求上的正整数；随后 Broker 会在与该订阅匹配的每一条 PUBLISH 上回传该标识符（§3.3.2.3.8），这样 Client 就可以将传入的消息路由到正确的 Handler，而无需重新解析 Topic。

它是作为 SUBSCRIBE 报文上的属性进行设置的，而不是在 PUBLISH 上：

MqttProperties subProps = new MqttProperties();
subProps.add(new MqttProperties.IntegerProperty(
    MqttProperties.MqttPropertyType.SUBSCRIPTION_IDENTIFIER.value(),
    42));

client.subscribe(java.util.Collections.singletonMap("sensors/+/temperature", 1), subProps);

当你需要将标识符与按 Topic 设置的 v5 订阅选项（No Local、Retain As Published、Retain Handling）结合使用时，可以将相同的 subProps 传递给基于 List 的重载 subscribe。

在接收端，从传入的 PUBLISH 中读取该标识符：

client.publishHandler(msg -> {
  MqttProperties.MqttProperty<?> p = msg.properties()
      .getProperty(MqttProperties.MqttPropertyType.SUBSCRIPTION_IDENTIFIER.value());
  if (p != null) {
    int id = (Integer) p.value(); // 42
    // 根据 id 进行路由
  }
});

如果 Broker 将该 PUBLISH 与多个订阅进行了匹配，该属性会重复出现 —— 使用 properties().getProperties(SUBSCRIPTION_IDENTIFIER.value()) 即可获取完整列表。

Client 所强制执行的前提条件（若尝试使用违背这些条件的 Subscription Identifier 进行订阅，返回的 Future 将失败并抛出 MqttException）：

  • 协议版本必须为 5（MqttClientOptions.setVersion(5)）—— 否则错误码为 MQTT_SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED；

  • Broker 不得在 CONNACK 中通告 SUBSCRIPTION_IDENTIFIER_AVAILABLE=0；协商后的值由 subscriptionIdentifierAvailable 暴露，缺失时按规范默认视为“可用”。

Topic 别名（Topic Alias）

Topic 别名（Topic Alias，MQTT 5.0 §3.3.2.3.4）在双向均由系统自动管理。

服务端 → 客户端： 当传入的 PUBLISH 携带 Topic Alias 属性时，别名到 Topic 的映射会被缓存，后续复用该别名且 Topic 名称为空的报文在交付给 publishHandler 之前会被透明地重新展开。Broker 可以使用的最大别名数由 setTopicAliasMaximum 控制；若别名值为 0 或高于协商的最大值，将导致连接断开并返回原因码 TOPIC_ALIAS_INVALID 的 DISCONNECT。

客户端 → 服务端： 当 Broker 在其 CONNACK 中通告了非零的 Topic Alias Maximum 时，Client 会透明地为发出的 PUBLISH 报文分配别名。针对给定 Topic 的首次 PUBLISH 携带完整的 Topic 名称以及新分配的 Topic Alias 属性；针对相同 Topic 的后续 PUBLISH 报文将发送空的 Topic 名称并仅携带别名。如果 Broker 通告的别名池耗尽，后续的 Topic 将使用其完整名称且不带别名进行发布。这完全是在内部进行的 —— 应用程序继续使用真实的 Topic 名称调用 publish，无需关注别名。

请求 / 响应交互模式

MQTT 5.0 §4.10 在普通 PUBLISH 报文的基础上标准化了请求/响应（Request/Response）模式，利用了三个新属性：Response Topic、Correlation Data 以及 Broker 可能通告的可选 Response Information 提示（以协助 Client 选择 Topic 前缀）。该 Client 完整支持此模式。

步骤 1 ——（可选）向 Broker 请求响应信息（Response Information）

如果你希望 Broker 建议一个 Topic 前缀作为 Response Topic（通常是该 Broker 针对该 Client 的 ACL 已经覆盖的每个 Client 专属的唯一样式路径），请在 CONNECT 上设置 Request Response Information = true：

MqttClientOptions options = new MqttClientOptions();
options.setVersion(io.netty.handler.codec.mqtt.MqttVersion.MQTT_5.protocolLevel());
options.setRequestResponseInformation(true);
MqttClient client = MqttClient.create(vertx, options);

client.connect(1883, "broker.example.com")
  .onSuccess(connack -> {
    String prefix = connack.responseInformation(); // 如果 Broker 没有返回任何内容则为 null
    // 例如 "$share/responses/{clientId}/"
  });

参阅 setRequestResponseInformation 与 responseInformation。Broker 可以自由选择忽略该请求 —— 此时将 null 视为“无建议，请自行选用响应主题”。

步骤 2 —— 请求方：携带 Response Topic 与 Correlation Data 发送 PUBLISH

请求方订阅其将用于接收回复的任意 Topic，然后发布设置了这两个属性的请求：

// 1. 订阅回复通道
String replyTopic = "clients/" + client.clientId() + "/replies";
client.subscribe(replyTopic, 1);

// 2. 发布请求
byte[] correlationId = java.util.UUID.randomUUID().toString().getBytes();

MqttProperties reqProps = new MqttProperties();
reqProps.add(new MqttProperties.StringProperty(
    MqttProperties.MqttPropertyType.RESPONSE_TOPIC.value(), replyTopic));
reqProps.add(new MqttProperties.BinaryProperty(
    MqttProperties.MqttPropertyType.CORRELATION_DATA.value(), correlationId));

client.publish("requests/temperature",
    Buffer.buffer("{\"room\":\"kitchen\"}"),
    MqttQoS.AT_LEAST_ONCE, false, false, reqProps);

Correlation Data 对 MQTT 而言是不透明的（opaque）—— 其唯一目的是当有多个请求同时在进行时，让请求方能够将回复与发起的请求进行配对。

步骤 3 —— 响应方：读取 Response Topic 与 Correlation Data 并回发消息

在响应方一侧，这两个属性随传入的 PUBLISH 一并到达：

client.publishHandler(msg -> {
  if (!"requests/temperature".equals(msg.topicName())) return;

  String respTopic = ((MqttProperties.StringProperty) msg.properties()
      .getProperty(MqttProperties.MqttPropertyType.RESPONSE_TOPIC.value())).value();
  byte[] corr = ((MqttProperties.BinaryProperty) msg.properties()
      .getProperty(MqttProperties.MqttPropertyType.CORRELATION_DATA.value())).value();

  MqttProperties respProps = new MqttProperties();
  respProps.add(new MqttProperties.BinaryProperty(
      MqttProperties.MqttPropertyType.CORRELATION_DATA.value(), corr));

  client.publish(respTopic,
      Buffer.buffer("{\"temp\":21.5}"),
      MqttQoS.AT_LEAST_ONCE, false, false, respProps);
});

无法或不愿意响应该请求的响应方（缺少 Response Topic、不受支持的 Payload 等）只需不发布回复即可 —— 该模式在协议层面并未定义否定确认（Negative Ack）。

步骤 4 —— 请求方：关联响应消息

回到请求方，同一个 publishHandler 回调在 replyTopic 上接收到回复。将 Correlation Data 与请求中使用的值进行匹配，以将响应路由给正确的调用者（例如完成保存在 Map<ByteBuffer, Promise<Buffer>> 中的 Promise<Buffer>）。

遗嘱消息与请求 / 响应

遗嘱消息也可以充当“我已离线”的回复：MqttClientWillOptions 暴露了 setResponseTopic(…​) 与 setCorrelationData(…​)，这样当发生异常断开时，Broker 将发布一条遗嘱消息，所有正在等待的请求方都可以据此进行关联。

服务端主动发起的断开连接（DISCONNECT）

可以注册一个 Handler，以便在服务端发送 DISCONNECT 报文时（而非 Client 自身主动关闭连接）接收通知：

client.disconnectMessageHandler(msg -> {
  System.out.println("server disconnected, reason=" + msg.code() + " props=" + msg.properties());
});

参阅 disconnectMessageHandler。该 Handler 在 closeHandler 之前触发，且仅针对服务端发起的断开连接触发。 Client 也可以通过 disconnect 主动发送携带原因码和属性的 DISCONNECT。

增强认证（Enhanced Authentication, AUTH）

MQTT 5.0 §4.12 引入了基于全新 AUTH 控制报文构建的 增强认证（Enhanced Authentication） 流程，旨在支持多轮质询/响应机制（例如 SCRAM、Kerberos、双向证明等），这些机制无法容纳在 3.1.1 所使用的单个 CONNECT 交互中。Client 端完全支持此流程，无论是在初始连接期间（§4.12.1），还是在已建立连接上的重新认证期间（§4.12.2）。

配置 CONNECT 报文

第一块认证数据以及认证机制的名称通过 MqttClientOptions 放置在 CONNECT 报文中：

  • setAuthenticationMethod —— 认证机制的 UTF-8 名称（MQTT 5.0 §3.1.2.11.9）。设置之后，Broker 应当通过 AUTH 报文推进后续的交互流程，如果不支持该机制，则必须以 Bad authentication method 拒绝该连接。

  • setAuthenticationData —— 所选认证机制所需的非透明二进制 Payload（MQTT 5.0 §3.1.2.11.10）。

Broker 在 CONNACK 中返回的值可以通过结果 MqttConnAckMessage 上的 authenticationMethod() 和 authenticationData() 进行查看 —— 对于诸如 SCRAM 之类的机制，最终的服务端签名（v=…​）通常会在该处交付。

处理来自 Broker 的 AUTH 报文

注册一个 Handler，以便在 Broker 发送 AUTH 报文时接收通知 —— 无论是在初始握手期间（在 CONNACK 之前，此时 Client 仍处于 CONNECTING 状态），还是在活动连接上的重新认证期间：

client.authenticationExchangeHandler(msg -> {
  // msg.reasonCode()           — CONTINUE_AUTHENTICATION / SUCCESS / RE_AUTHENTICATE
  // msg.authenticationMethod() — 由 Broker 回传，必须与协商好的方法匹配
  // msg.authenticationData()   — 认证机制的非透明 Payload（例如 SCRAM server-first-message）
});

参阅 authenticationExchangeHandler 与 MqttAuthenticationExchangeMessage。

发送 AUTH 报文

Client 通过 authenticationExchange 回复（或发起）AUTH 交互。它在两个阶段有效：

  • 初始增强认证（MQTT 5.0 §4.12.1）—— 当 client.connect(…​) 仍在进行中时，在 authenticationExchangeHandler 内部调用它。回复 CONTINUE_AUTHENTICATION（0x18），直到 Broker 以 CONNACK 终止交互。一旦 AUTH 帧被发送至网络，返回的 Future 就会成功；整体结果由 client.connect(…​) 交付。

  • 重新认证（MQTT 5.0 §4.12.2）—— 一旦连接建立，调用它并传入 RE_AUTHENTICATE（0x19）以重新启动认证交互，而无需关闭底层 TCP 连接。

原因码通过 MqttAuthenticateReasonCode 提供（SUCCESS、CONTINUE_AUTHENTICATION、RE_AUTHENTICATE）；MqttProperties 参数必须包含 AUTHENTICATION_METHOD（在整个交互过程中原样回显），并在机制需要时包含 AUTHENTICATION_DATA Payload。

示例：SCRAM-SHA-256

以下代码片段展示了如何针对通告 SCRAM-SHA-256 的 MQTT 5.0 Broker（例如配置了 scram 认证器的 EMQX）执行完整的 SCRAM-SHA-256 握手（RFC 5802）。加密原语均来自 JDK —— 无需额外依赖。

步骤 1 —— 生成客户端 Nonce，构建 client-first-message 并将其放置在 CONNECT 报文中：

String clientNonce = randomBase64(24);                       // 安全随机，RFC 5802 §5.1
String clientFirstMessageBare = "n=" + username + ",r=" + clientNonce;
String clientFirstMessage = "n,," + clientFirstMessageBare;  // "n,," 为 GS2 头

MqttClientOptions options = new MqttClientOptions();
options.setVersion(io.netty.handler.codec.mqtt.MqttVersion.MQTT_5.protocolLevel());
options.setAuthenticationMethod("SCRAM-SHA-256");
options.setAuthenticationData(Buffer.buffer(clientFirstMessage.getBytes(StandardCharsets.UTF_8)));
MqttClient client = MqttClient.create(vertx, options);

步骤 2 —— 当 Broker 以 server-first-message 进行回复时，计算客户端证明并通过 AUTH 报文回发 client-final-message：

client.authenticationExchangeHandler(msg -> {
  // server-first-message: r=<server-nonce>,s=<base64-salt>,i=<iterations>
  String serverFirst = new String(msg.authenticationData().getBytes(), StandardCharsets.UTF_8);
  Map<String,String> f = parseScram(serverFirst);
  String serverNonce = f.get("r");
  byte[] salt        = Base64.getDecoder().decode(f.get("s"));
  int iterations     = Integer.parseInt(f.get("i"));

  // RFC 5802 §5.1: 服务端必须扩展客户端 nonce —— 否则遵循安全失败（fail closed）原则抛出异常。
  if (!serverNonce.startsWith(clientNonce)) throw new SecurityException("bad nonce");

  String clientFinalNoProof = "c=biws,r=" + serverNonce;     // biws = base64("n,,")
  String authMessage = clientFirstMessageBare + "," + serverFirst + "," + clientFinalNoProof;

  byte[] saltedPwd    = pbkdf2(password, salt, iterations);          // PBKDF2-HMAC-SHA-256
  byte[] clientKey    = hmacSha256(saltedPwd, "Client Key".getBytes());
  byte[] storedKey    = sha256(clientKey);
  byte[] clientSig    = hmacSha256(storedKey, authMessage.getBytes());
  byte[] clientProof  = xor(clientKey, clientSig);

  String clientFinal = clientFinalNoProof
      + ",p=" + Base64.getEncoder().encodeToString(clientProof);

  MqttProperties props = new MqttProperties();
  props.add(new MqttProperties.StringProperty(
      MqttProperties.MqttPropertyType.AUTHENTICATION_METHOD.value(), "SCRAM-SHA-256"));
  props.add(new MqttProperties.BinaryProperty(
      MqttProperties.MqttPropertyType.AUTHENTICATION_DATA.value(),
      clientFinal.getBytes(StandardCharsets.UTF_8)));

  client.authenticationExchange(MqttAuthenticateReasonCode.CONTINUE_AUTHENTICATION, props);
});

client.connect(1883, "broker.example.com").onSuccess(connack -> {
  // 认证成功；connack.authenticationData() 可能携带 SCRAM server-final-message (v=...)
});

包含 SCRAM 辅助方法（pbkdf2、hmacSha256、sha256、xor、nonce 生成以及 SASL 名称转义）的完整可运行版本可参阅 public class VertxMqttClientAUTHExamples {

private static final String AUTH_METHOD = "SCRAM-SHA-256";
private static final String GS2_HEADER = "n,,";
public static void main(String[] args) {
final String host = args.length > 0 ? args[0] : "localhost";
final int port = args.length > 1 ? Integer.parseInt(args[1]) : 1883;
final String username = args.length > 2 ? args[2] : "user";
final String password = args.length > 3 ? args[3] : "public";
final String clientNonce = generateNonce();
final String clientFirstMessageBare = "n=" + saslName(username) + ",r=" + clientNonce;
final byte[] clientFirstMessage = (GS2_HEADER + clientFirstMessageBare).getBytes(StandardCharsets.UTF_8);
MqttClientOptions option = new MqttClientOptions();
option.setVersion(5);
option.setAuthenticationMethod(AUTH_METHOD);
option.setAuthenticationData(Buffer.buffer(clientFirstMessage));
Vertx vertx = Vertx.vertx();
MqttClient client = MqttClient.create(vertx, option);
client.authenticationExchangeHandler(msg -> {
  try {
    String serverFirstMessage = new String(msg.authenticationData().getBytes(), StandardCharsets.UTF_8);
    Map<String, String> fields = parseScramMessage(serverFirstMessage);
    String serverNonce = fields.get("r");
    byte[] salt = Base64.getDecoder().decode(fields.get("s"));
    int iterations = Integer.parseInt(fields.get("i"));
// RFC 5802 §5.1: 服务端必须扩展客户端 nonce —— 否则遵循安全失败（fail closed）原则抛出异常。
if (serverNonce == null || !serverNonce.startsWith(clientNonce)) {
  throw new SecurityException("Invalid server nonce");
}
String channelBinding = Base64.getEncoder().encodeToString(GS2_HEADER.getBytes(StandardCharsets.UTF_8));
String clientFinalNoProof = "c=" + channelBinding + ",r=" + serverNonce;
String authMessage = clientFirstMessageBare + "," + serverFirstMessage + "," + clientFinalNoProof;
byte[] saltedPassword = pbkdf2(password, salt, iterations);
byte[] clientKey = hmacSha256(saltedPassword, "Client Key".getBytes(StandardCharsets.UTF_8));
byte[] storedKey = MessageDigest.getInstance("SHA-256").digest(clientKey);
byte[] clientSignature = hmacSha256(storedKey, authMessage.getBytes(StandardCharsets.UTF_8));
byte[] clientProof = xor(clientKey, clientSignature);
String clientFinalMessage = clientFinalNoProof
  + ",p=" + Base64.getEncoder().encodeToString(clientProof);
MqttProperties props = new MqttProperties();
props.add(new MqttProperties.StringProperty(AUTHENTICATION_METHOD.value(), AUTH_METHOD));
props.add(new MqttProperties.BinaryProperty(AUTHENTICATION_DATA.value(),
  clientFinalMessage.getBytes(StandardCharsets.UTF_8)));
    client.authenticationExchange(MqttAuthenticateReasonCode.CONTINUE_AUTHENTICATION, props);
  } catch (Exception e) {
    e.printStackTrace();
  }
});
  client.connect(port, host).onComplete(ar -> {
    if (ar.succeeded()) {
      System.out.println("Connected & authenticated as '" + username + "'");
      client.publish("temperature", Buffer.buffer("hello"), MqttQoS.AT_LEAST_ONCE, false, false)
        .onComplete(p -> client.disconnect().onComplete(d -> vertx.close()));
    } else {
      System.err.println("Connect failed: " + ar.cause());
      vertx.close();
    }
  });
}
private static String generateNonce() {
  byte[] bytes = new byte[24];
  new SecureRandom().nextBytes(bytes);
  // RFC 5802 §5.1: nonce 必须是“可打印字符” —— 剔除 SCRAM 属性中不允许的 base64 字符。
  return Base64.getEncoder().encodeToString(bytes).replace("=", "").replace(",", "");
}
private static String saslName(String name) {
  return name.replace("=", "=3D").replace(",", "=2C");
}
private static Map<String, String> parseScramMessage(String msg) {
  Map<String, String> out = new HashMap<>();
  for (String token : msg.split(",")) {
    int eq = token.indexOf('=');
    if (eq > 0) {
      out.put(token.substring(0, eq), token.substring(eq + 1));
    }
  }
  return out;
}
private static byte[] pbkdf2(String password, byte[] salt, int iterations) throws Exception {
  SecretKeyFactory skf = SecretKeyFactory.getInstance("PBKDF2WithHmacSHA256");
  PBEKeySpec spec = new PBEKeySpec(password.toCharArray(), salt, iterations, 256);
  try {
    return skf.generateSecret(spec).getEncoded();
  } finally {
    spec.clearPassword();
  }
}
private static byte[] hmacSha256(byte[] key, byte[] data) throws Exception {
  Mac mac = Mac.getInstance("HmacSHA256");
  mac.init(new SecretKeySpec(key, "HmacSHA256"));
  return mac.doFinal(data);
}
  private static byte[] xor(byte[] a, byte[] b) {
    byte[] out = new byte[a.length];
    for (int i = 0; i < a.length; i++) {
      out[i] = (byte) (a[i] ^ b[i]);
    }
    return out;
  }
}.

服务端 AUTH 支持

在服务端，AUTH 报文在传输层面上是被接收的（因此可以在不断开 TCP 连接的情况下探测 Broker），但 MqttEndpointImpl 目前尚未将其暴露给用户代码：handleAuth 仅为一个桩（stub）。因此，实现增强认证 Endpoint 需要扩展 Endpoint API；客户端类型已经就绪，因此未来的服务端支持不会改变本节的 API 表面。

自动服务端重定向

当 Broker 返回携带 Server Reference 属性的 CONNACK 或 DISCONNECT（MQTT 5.0 §3.2.2.3.18 / §3.14.2.3.4）时，Client 可以透明地重连到此处所列出的其中一台服务器。该行为默认开启，可通过 setAutoServerRedirect 进行开关。当逗号分隔的列表中存在多个服务器引用时，将随机选取其中一个。