RabbitMQ作为MQTT Broker时客户端重复接收保留消息求助

核心问题 #

用RabbitMQ做MQTT Broker时，服务器端发送retain=true的消息，Android客户端断开重连后，不仅收到最新消息，还会收到更早的历史消息；但换用ActiveMQ时完全正常。

排查与解决步骤 #

1. 确认RabbitMQ保留消息的实际存储 #

RabbitMQ严格遵循MQTT标准，每个主题只会保留最新一条retain消息。先通过RabbitMQ管理后台（默认15672端口）的MQTT标签页，查看Retained Messages列表：

• 如果目标主题存在多条保留消息，手动清理后重新测试重连行为。
• 清理后问题消失的话，说明之前的发布逻辑可能触发了多次成功发布，导致重复写入保留消息。

2. 修正客户端订阅逻辑 #

你的Android客户端在connectComplete中每次连接都会调用subscribeTopic，虽设置了cleanSession=true，但仍可能因RabbitMQ会话清理不及时导致重复订阅，进而接收重复消息。修改订阅逻辑，先取消旧订阅再重新订阅：

private static void subscribeTopic() {
    try {
        // 先取消已有订阅，避免重复
        mqttAndroidClient.unsubscribe(getDownstreamTopic());
        mqttAndroidClient.subscribe(getDownstreamTopic(), 1, activity, new IMqttActionListener() {
            @Override
            public void onSuccess(IMqttToken asyncActionToken) {
                LOGGER.debug("subscribed succeed :{}", getDownstreamTopic());
            }

            @Override
            public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                LOGGER.warn("subscribed failed", exception);
            }
        });
    } catch (MqttException e) {
        LOGGER.warn("subscribe failed", e);
    }
}

3. 调整RabbitMQ MQTT插件配置 #

检查rabbitmq.conf中的以下配置，确保符合预期：

• 启用保留消息持久化（默认开启，若被修改需恢复）：

  mqtt.retain_persistence = true
  

• 关闭重复消息允许（避免因网络波动导致的重复投递）：

  mqtt.allow_duplicate_messages = false
  

修改配置后重启RabbitMQ服务。

4. 优化发布重试逻辑 #

你的MqttTask在发布失败时会立即重试，短时间内重复发送retain消息，可能触发RabbitMQ的极端场景异常。给重试添加延迟，降低重复发布的概率：

if (count < MAX_RETRY) {
    // 延迟1秒后重试，避免短时间内重复发送
    sendExecutor.schedule(new MqttTask(this.count + 1, this.mqttTaskTopic, this.mqttTaskMessage), 1, TimeUnit.SECONDS);
}

5. 调整客户端连接参数 #

• 服务器端生产者：将cleanSession设为false，确保生产者会话持久化，避免重复连接导致的异常发布：

  conOpt.setCleanSession(false);
  

• Android客户端：订阅时开启noLocal参数，避免收到客户端自身发布的消息（如果存在交叉发布场景）：

  // 使用带noLocal参数的重载订阅方法
  mqttAndroidClient.subscribe(getDownstreamTopic(), 1, null, true, new IMqttMessageListener() {
      @Override
      public void messageArrived(String topic, MqttMessage message) throws Exception {
          processPayload(topic, message);
      }
  });
  

总结 #

大概率是RabbitMQ插件配置或客户端重复订阅导致的问题，按上述步骤逐一排查，先从管理后台确认保留消息数量，再修正订阅和发布逻辑，最后调整配置即可解决。

内容的提问来源于stack exchange，提问作者Arrow.Tao