mqtt客户端断联后 订阅消失

EMQX
1

mqtt断开7秒钟，订阅全部消失，这是为啥
客户端设置了记住客户端状态，会话过期时间为120
我尝试重新订阅 也无法订阅
options.setWill(“WillTopic”, message);
|||options.setCleanStart(false);
|||options.setSessionExpiryInterval(120L);
|||options.setKeepAliveInterval(MqttConfig.getKeepAliveInterval());
|||options.setConnectionTimeout(MqttConfig.getTimeout());
|||options.setMaxReconnectDelay(60);
|||options.setServerURIs(new String { MqttConfig.getBroker() });
|||options.setAutomaticReconnect(true);
|||options.setAutomaticReconnectDelay(10, 60);

image
image1845×115 97.5 KB

image
image685×735 50.6 KB

bf8508bd-7e34-4d0e-965d-5d1c2ca1b51a
bf8508bd-7e34-4d0e-965d-5d1c2ca1b51a1920×922 71.8 KB

2

订阅大概率没丢，是会话恢复后你又用同一批 Subscription Identifier 重新订阅了。
截图里重连后 Clean Start=false，订阅数还有 425；日志里的 The Subscription Identifier ... already exists 也说明 EMQX 侧还保存着原订阅/订阅标识，不是“断 7 秒后清空”。
你现在是 MQTT v5、Clean Start=false、Session Expiry=120，同一个 Client ID 在 120 秒内重连会恢复原会话。这时不要无条件重新订阅。先看 CONNACK 的 Session Present：为 true 就跳过全量订阅；为 false 才订阅。

客户端逻辑改成这样：

options.setCleanStart(false);
options.setSessionExpiryInterval(120L);
// reconnect/connect success 后
if (!connAckSessionPresent) {
    subscribeAll();
}

如果当前 Java 库拿不到 Session Present，临时规避是重连后不要带 Subscription Identifier 重新订阅，或者每次给每个订阅生成新的 identifier，别复用已经存在的那批 ID。

Dashboard 上断线时看不到在线客户端/订阅，不能直接当成会话被删。用同一个 clientId 重连后，看连接详情里的 Clean Start=false、会话是否 resumed，以及消息是否还能按原 topic 下发。

3

我之前是在连接后 进行订阅，重连不订阅，但是订阅还是没了
@Override
public void connectComplete(boolean reconnect, String serverURI) {
log.info("连接完成 " + serverURI + " " + reconnect);
if (!reconnect) {
// 订阅
initSubscribe();
}
}

那么理论上来说 这么写是对的

如果我不保留会话，使用同一clientId 重连 再订阅 也是可以的吧

4

如果订阅没有丢，我重新订阅报错了，原来的订阅也会消失吗？
现在的情况是 订阅确实没了

5

connectComplete(boolean reconnect, ...) 不能当成“会话已恢复”的判断。
reconnect=true 只说明这是客户端库自动重连，不等于 EMQX 返回了旧会话。MQTT 5 要看 CONNACK 里的 Session Present：

  • Session Present=true：旧会话还在，订阅应还在，可以跳过全量订阅。
  • Session Present=false：服务端没有恢复旧会话，这次是新会话，必须重新订阅。
    所以逻辑不要只写成 if (!reconnect) subscribe()。如果客户端库拿不到 Session Present，先用最稳的方式处理：每次连接成功都订阅一遍，但不要复用固定的 Subscription Identifier。
@Override
public void connectComplete(boolean reconnect, String serverURI) {
    log.info("连接完成 {} reconnect={}", serverURI, reconnect);
    initSubscribe(); // 先去掉固定 Subscription Identifier，或保证每个订阅 ID 唯一
}

同一个 clientId 重连后再订阅是可以的。已有同一 topic filter 时，SUBSCRIBE 应该是覆盖/更新订阅参数，不应该把原订阅删掉。Subscription Identifier already exists 说明这次 SUBSCRIBE 被拒了，正常情况下不会因为这次失败把旧订阅清空。
如果 Dashboard 里确实变成 0 个订阅，优先查这几项：

  1. 重连前后 clientId 是否完全一样；
  2. CONNACK 里的 Session Present 是 true 还是 false；
  3. 客户端实际发出的 CONNECT 里 Clean Start 和 Session Expiry Interval 是多少；
  4. initSubscribe() 里是否给不同 topic 复用了固定 Subscription Identifier；
  5. 是否有同 clientId 的第二个连接接管了会话。
    先把重连那次 CONNECT/CONNACK 日志、SUBSCRIBE reason code、initSubscribe() 里设置 Subscription Identifier 的代码贴出来。现在看更像是“会话没有被恢复，但代码因为 reconnect=true 跳过了订阅”，或者“订阅时复用了固定 Subscription Identifier 导致 SUBSCRIBE 失败”，不是 EMQX 主动把已有订阅删了。
6

1 clientId 完全一样
2 不清楚
3
|||options.setCleanStart(false);
|||options.setSessionExpiryInterval(120L);
4 不同 topic 复用了固定 Subscription Identifier
MqttProperties mqttProperties = new MqttProperties();
List subscriptionIdentifiers = new ArrayList<>();
Integer property = (mqttSubscription.getTopic()).hashCode();
property = Math.abs(property) % 268435455;
subscriptionIdentifiers.add(property);
mqttProperties.setSubscriptionIdentifiers(subscriptionIdentifiers);
client.subscribe(mqttSubscription, “”, null, mqttTopicListener, mqttProperties);
5 不清楚

这个是当时的现场日志

image
image1845×115 97.5 KB

从这看 重连之后进行了订阅，之前是这样写的
@Override
public void connectComplete(boolean reconnect, String serverURI) {
log.info("连接完成 " + serverURI + " " + reconnect);
// 订阅
initSubscribe();
}

7

基本明确了：旧会话恢复后，原来的订阅还在，但代码又用同一个 Subscription Identifier 重新订了一遍，所以 EMQX 拒绝了这次 SUBSCRIBE。

你这里不是“不同 topic 共用同一个 ID”这么简单，而是每个 topic 用 hashCode() 算出固定 ID。会话恢复后，服务端还保存着这个 topic 的旧订阅和旧 Subscription Identifier；重连后再次订阅同一个 topic、带同一个 ID，就会触发日志里的：

The Subscription Identifier ... already exists

先按这个方式改，别先怀疑 EMQX 把订阅删了：

// 临时排障：先不要设置 Subscription Identifier
// mqttProperties.setSubscriptionIdentifiers(subscriptionIdentifiers);
client.subscribe(mqttSubscription, "", null, mqttTopicListener, mqttProperties);

如果业务确实需要 Subscription Identifier，就不要用固定 topic hash 直接复用。要么在确认 Session Present=false 时才全量订阅，要么给“本次新建的订阅”分配当前会话内没被占用的 ID。Session Present=true 时，旧订阅本来就应该继续用，不要重复全量订阅。

另外，SUBSCRIBE 被 Subscription Identifier already exists 拒绝，正常不会主动删除旧订阅。如果 Dashboard 里看到订阅数变 0，继续查这三项：

  1. CONNACK 里的 Session Present 到底是 true 还是 false；
  2. 有没有第二个相同 clientId 的连接把会话顶掉；
  3. 断线和重连是否落到不同节点，且旧会话没有被恢复。
    先把 Session Present 打出来；如果客户端库暂时拿不到，就先去掉 Subscription Identifier 跑一轮验证。