通过WebSocket使用MQTT时RabbitMQ管理端无法查看队列消息

按照RabbitMQ文档指引，在浏览器中通过MQTT over WebSocket订阅/topic/test主题接收消息，同时使用mqtt-explorer工具向RabbitMQ发送消息。浏览器控制台可正常接收消息，但RabbitMQ管理门户仅能看到自动生成的队列，且队列消息数始终为0；当浏览器未连接时，发送的消息也不会存入RabbitMQ队列。

浏览器端代码如下：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Hello MQTT</title>
</head>
<body>
    <h1>Hello MQTT</h1>
    <p>This is a basic HTML page with the title</p>

    <!-- Source library -->
    <script src="mqttws31.js" type="text/javascript"></script>

    <script>

        var wsbroker = "localhost"
        var wsport = 15675; // port for above
        var client = new Paho.MQTT.Client(wsbroker, wsport, "/ws",
            "myclientid_" + parseInt(Math.random() * 100, 10));

        client.onConnectionLost = function (responseObject) {
            console.log("CONNECTION LOST - " + responseObject.errorMessage);
        };

        client.onMessageArrived = function (message) {
            console.log("RECEIVE ON " + message.destinationName + " PAYLOAD " + message.payloadString);
        };

        var options = {
            timeout: 3,
            keepAliveInterval: 30,

            onSuccess: function () {
                console.log("CONNECTION SUCCESS");
                client.subscribe('/topic/test', {qos: 1});
            },

            onFailure: function (message) {
                console.log("CONNECTION FAILURE - " + message.errorMessage);
            }
        };
        
        console.log("CONNECT TO " + wsbroker + ":" + wsport);
        client.connect(options);

    </script>
</body>
</html>

问题根源 #

RabbitMQ的MQTT插件默认会为每个客户端创建临时排他队列，这类队列有两个核心特性：

1. 客户端断开连接后自动删除
2. 消息被消费后立即从队列中移除，不会留存

这就导致了管理门户中队列消息数始终为0，且离线时消息无法存储的问题。

具体调整步骤 #

1. 修改客户端连接配置，禁用干净会话
   在浏览器端的连接选项中添加cleanSession: false，让RabbitMQ保留客户端的订阅关系和对应的持久化队列：

   var options = {
       timeout: 3,
       keepAliveInterval: 30,
       cleanSession: false, // 关键配置：保留会话和队列
       onSuccess: function () {
           console.log("CONNECTION SUCCESS");
           client.subscribe('/topic/test', {qos: 1});
       },
       onFailure: function (message) {
           console.log("CONNECTION FAILURE - " + message.errorMessage);
       }
   };
   

2. 发布消息时设置持久化属性
   使用mqtt-explorer发送消息时，需配置以下参数：

   • 将QoS设置为1或2
   • 勾选Retain选项（让RabbitMQ留存最后一条消息，供离线客户端上线后接收）
   • 开启消息持久化（部分工具中对应“Persistent”选项）

3. 验证队列属性（可选）
   在RabbitMQ管理门户中检查队列属性，确保：

   • Durable为Yes（队列持久化，重启服务不丢失）
   • Exclusive为No（非排他队列，允许多客户端访问）
   • Auto-delete为No（客户端断开后队列不自动删除）

   若默认生成的队列不符合要求，可手动创建持久化队列，并将/topic/test主题绑定到该队列，路由键设置为/topic/test。

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