HiveMQ CE客户端与Broker保活机制异常问题排查求助

背景信息 #

使用hivemq-mqtt-client-1.3.0.jar构建MQTT5客户端，客户端设置60秒KeepAlive时间，连接代码如下：

String id = "1";
Mqtt5ClientBuilder builder = Mqtt5Client.builder()
                .identifier(id)
                .serverHost("localhost")
                .serverPort(1883)
                .addConnectedListener(context -> System.out.println("Connected Now"))
                .addDisconnectedListener(context -> System.out.println("Disconnected Connected Now"));
        
client = builder.build();
client.toBlocking().connectWith().cleanStart(false).keepAlive(keepAlive).send();
subscribe(id, "1");

连接Broker后，event.txt日志显示：

2023-01-03 12:56:39,428 - Client ID: 1, IP: 127.0.0.1, Clean Start: false, Session Expiry: 0 connected.

场景一：主动断开后Broker未保留会话 #

操作与问题 #

客户端连接10秒后调用主动断开代码：

Thread.sleep(10*1000);
client.toBlocking().disconnect();

预期Broker基于60秒KeepAlive，断开后保留会话50秒，但Broker立刻输出断开日志：

2023-01-03 12:56:49,482 - Client ID: 1, IP: 127.0.0.1 disconnected gracefully.

错误分析 #

1. 概念混淆：KeepAlive是连接存活检测机制，用于Broker判断客户端是否离线，和会话保留时长无关。会话保留由Session Expiry Interval控制，从连接日志可见Session Expiry: 0，意味着客户端断开后会话立即失效，不会保留。
2. 主动断开的行为：调用disconnect()是优雅断开流程，客户端会发送DISCONNECT报文给Broker，Broker收到后会立即终止连接并清理会话（因会话过期时间为0），不会等待KeepAlive超时。

解决建议 #

• 若需保留会话，连接时设置Session Expiry Interval，示例设置为3600秒（1小时）：

  client.toBlocking().connectWith()
          .cleanStart(false)
          .keepAlive(60)
          .sessionExpiryInterval(3600) // 添加会话过期时间配置
          .send();
  

• 明确KeepAlive与Session Expiry的区别：KeepAlive是Broker检测客户端在线状态的超时阈值（1.5倍时长未收到报文则判定离线），Session Expiry是客户端断开后Broker保留会话的时长。

场景二：订阅后无法接收消息，KeepAlive机制未触发 #

操作与问题 #

不主动调用disconnect，仅执行订阅代码：

public static void subscribe(String id, String cid) {
    try {
        
        client.toAsync().subscribeWith()
                .topicFilter("TEMP/"+id+"/"+cid).callback(publish -> {
                    
                    String message = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
                    System.out.println("Received message on topic " + publish.getTopic() + ": " + message);

                }).send();
        
        System.out.println("Subscribing to " + "TEMP/"+id+"/"+cid);

出现两个问题：

1. 无法接收TEMP/1/1主题的消息
2. Broker未在90秒（60*1.5）后断开客户端，也未发送PINGREQ

错误分析 #

无法接收消息的原因：

1. 代码不完整：提供的订阅代码缺少catch/finally块，可能因未捕获异常导致订阅操作未完成。
2. 异步订阅未确认：使用toAsync().subscribeWith().send()是异步操作，未等待订阅确认就结束主线程，客户端进程退出后无法接收消息。
3. 主题匹配问题：发布消息的主题可能与订阅的TEMP/1/1不匹配（MQTT主题区分大小写、层级分隔符为/）。

KeepAlive机制未触发的原因：

• 客户端主线程已退出：若主程序在订阅后没有阻塞逻辑，客户端进程会自动关闭，Broker不会触发KeepAlive检测，客户端也不会发送PINGREQ。

解决建议 #

修复订阅问题：

1. 补全订阅代码的异常处理，并添加订阅成功确认回调：

   public static void subscribe(String id, String cid) {
       try {
           client.toAsync().subscribeWith()
                   .topicFilter("TEMP/" + id + "/" + cid)
                   .callback(publish -> {
                       String message = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
                       System.out.println("Received message on topic " + publish.getTopic() + ": " + message);
                   })
                   .send()
                   .whenComplete((subAck, throwable) -> {
                       if (throwable != null) {
                           System.err.println("订阅失败：" + throwable.getMessage());
                       } else {
                           System.out.println("订阅成功，返回码：" + subAck.getReasonCodes());
                       }
                   });
           System.out.println("发起订阅请求：" + "TEMP/" + id + "/" + cid);
       } catch (Exception e) {
           e.printStackTrace();
       }
   }
   

2. 保持主线程运行：在主程序末尾添加阻塞逻辑，避免客户端进程退出：

   public static void main(String[] args) throws InterruptedException {
       // 连接、订阅代码执行完成后
       Thread.sleep(Long.MAX_VALUE); // 阻塞主线程
   }
   

3. 验证主题匹配：确保发布消息的主题与TEMP/1/1完全一致，检查大小写、层级是否正确。

修复KeepAlive检测问题：

• 确保客户端进程持续运行，HiveMQ客户端会自动在KeepAlive周期的一半时间发送PINGREQ（如60秒KeepAlive，30秒时自动发送），无需手动处理。
• 当前Broker配置的max-keep-alive为60，与客户端设置一致，无需修改；若需Broker强制断开超时客户端，确认无其他配置覆盖KeepAlive检测逻辑。

内容的提问来源于stack exchange，提问作者YDev01