HTTP 輪詢 vs MQTT：SpringBoot 通信實踐

作者：一安 2025-09-26 07:12:04

為優化普通輪詢的資源浪費問題，長輪詢對邏輯進行了調整：客戶端發送請求后，服務端不會立即返回響應，而是掛起請求（通常設置超時時間，如30秒）；若期間有新數據，服務端立即返回響應；客戶端收到響應后，立即發起下一次長輪詢。

引言

在實時通信場景中，消息傳遞的效率、可靠性與資源占用一直是開發者關注的核心。從早期的HTTP輪詢到如今廣泛應用的MQTT協議，技術方案的演進始終圍繞更高效地實現端到端通信這一目標展開。

技術演進：為什么從 HTTP 輪詢走向 MQTT？

HTTP 輪詢：簡單但低效的被動通信

HTTP 協議作為互聯網的基礎協議，基于請求 - 響應模型設計，天然適合客戶端主動發起請求、服務端被動返回數據的場景。但在實時通信（如即時聊天、設備狀態監控、消息推送）中，為了獲取實時更新的數據，開發者不得不采用輪詢方案，常見的實現方式有兩種：

（1）普通輪詢（Polling）

客戶端按照固定時間間隔（如1秒、5秒）向服務端發送HTTP請求，查詢是否有新數據；服務端無論是否有數據，都會立即返回響應。

核心問題：

資源浪費嚴重：大部分請求是無效請求（服務端無新數據），卻占用了網絡帶寬、服務端連接數與 CPU 資源；

實時性差：數據更新的延遲等于輪詢間隔（如5秒輪詢，延遲最高可達5秒），無法滿足低延遲場景需求。

（2）長輪詢（Long Polling）

核心問題：

連接占用時間長：服務端需要維護大量掛起的HTTP連接，在高并發場景下會消耗大量內存與線程資源；

協議開銷大：HTTP請求頭（如Cookie、User-Agent）通常占整個請求體積的70%以上，即使僅傳遞少量數據，也需要攜帶完整的請求頭，帶寬利用率低；

不支持多對多通信：HTTP輪詢本質是客戶端 - 服務端的點對點通信，無法直接實現設備間、客戶端間的消息轉發。

MQTT：為實時、低耗通信而生

MQTT（Message Queuing Telemetry Transport，消息隊列遙測傳輸）是1999年誕生的輕量級發布 / 訂閱（Pub/Sub）協議，最初用于石油管道監控場景，如今已成為物聯網（IoT）、實時消息推送的主流協議。其設計目標是在帶寬有限、網絡不穩定的環境下，實現可靠的低功耗通信，核心特性完美解決了HTTP輪詢的痛點：

特性	說明
發布 / 訂閱模型	客戶端（發布者）不直接與接收者（訂閱者）通信，而是通過 “主題（Topic）” 轉發消息，支持多對多通信；
輕量級協議	協議頭最小僅 2 字節，遠低于 HTTP 的幾十 KB，帶寬利用率極高；
持久化連接	客戶端與服務端建立一次 TCP 連接后，可長期復用，無需頻繁建立連接，減少資源消耗；
QoS 服務質量	支持 3 級消息可靠性：QoS 0（最多一次）、QoS 1（至少一次）、QoS 2（恰好一次）；
斷開重連與遺囑	客戶端異常斷開時，服務端可自動觸發遺囑消息（Last Will and Testament），通知其他訂閱者；
消息保留	服務端可保留某個主題的最新消息，新訂閱者上線后可直接獲取該消息，無需等待發布者再次發送；

實踐案例

核心組件：

MQTT服務端（Broker）：負責接收、存儲、轉發消息，常見實現有Eclipse Mosquitto（開源輕量）、EMQX（企業級）、AWS IoT Core等；
MQTT客戶端：SpringBoot應用作為客戶端，實現發布消息與訂閱消息功能，常用客戶端庫為Eclipse Paho。

集成 MQTT 客戶端

引入依賴

<!-- MQTT客戶端 -->
<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

配置 MQTT 連接參數

spring:
  mqtt:
    # MQTT服務端地址（tcp://ip:端口）
    broker-url: tcp://localhost:1883
    # 客戶端ID（必須唯一，建議添加隨機后綴避免沖突）
    client-id: springboot-mqtt-client-${random.uuid}
    # 用戶名（Mosquitto默認無密碼，若配置了認證需填寫）
    username: 
    # 密碼
    password: 
    # 默認訂閱的主題（可配置多個，用逗號分隔）
    default-topics: test/topic, device/status
    # QoS級別（0/1/2）
    qos: 1
    # 是否自動重連
    automatic-reconnect: true
    # 連接超時時間（毫秒）
    connection-timeout: 3000
    # 保持連接心跳時間（秒）
    keep-alive-interval: 60

編寫 MQTT 配置類：初始化客戶端

@Configuration
@ConfigurationProperties(prefix = "spring.mqtt") // 綁定application.yml中的配置
@Data
public class MqttConfig {

    private String brokerUrl;
    private String clientId;
    private String username;
    private String password;
    private String[] defaultTopics;
    private int qos;
    private boolean automaticReconnect;
    private int connectionTimeout;
    private int keepAliveInterval;

    /**
     * 初始化MQTT客戶端
     */
    @Bean
    public MqttClient mqttClient() throws MqttException {
        // 1. 創建連接選項
        MqttConnectOptions options = new MqttConnectOptions();
        // 設置用戶名密碼（若服務端無認證，可省略）
        if (username != null && !username.isEmpty()) {
            options.setUserName(username);
        }
        if (password != null && !password.isEmpty()) {
            options.setPassword(password.toCharArray());
        }
        // 設置自動重連、連接超時、心跳時間
        options.setAutomaticReconnect(automaticReconnect);
        options.setConnectionTimeout(connectionTimeout);
        options.setKeepAliveInterval(keepAliveInterval);
        // 禁用“清除會話”（確保斷開重連后，未接收的消息能繼續接收）
        options.setCleanSession(false);

        // 2. 創建MqttClient實例（MemoryPersistence表示消息持久化到內存）
        MqttClient client = new MqttClient(brokerUrl, clientId, new MemoryPersistence());

        // 3. 設置客戶端回調（處理連接成功、消息到達、連接丟失等事件）
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                // 連接丟失時觸發（可在這里實現重連邏輯，不過options已配置自動重連）
                System.out.println("MQTT連接丟失：" + cause.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                // 收到訂閱的消息時觸發
                System.out.println("收到MQTT消息：");
                System.out.println("主題：" + topic);
                System.out.println("內容：" + new String(message.getPayload()));
                System.out.println("QoS級別：" + message.getQos());
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                // 消息發布完成時觸發（僅QoS>0時有效）
                try {
                    if (token.isComplete()) {
                        System.out.println("消息發布成功：" + token.getMessageId());
                    }
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        });

        // 4. 連接服務端并訂閱默認主題
        client.connect(options);
        if (defaultTopics != null && defaultTopics.length > 0) {
            // 訂閱多個主題（第二個參數為QoS數組，與主題數組一一對應）
            int[] qosArray = new int[defaultTopics.length];
            for (int i = 0; i < defaultTopics.length; i++) {
                qosArray[i] = qos;
            }
            client.subscribe(defaultTopics, qosArray);
            System.out.println("MQTT連接成功，已訂閱主題：" + String.join(",", defaultTopics));
        }

        return client;
    }
}

編寫 MQTT 工具類：封裝發布 / 訂閱方法

@Component
public class MqttUtil {

    @Resource
    private MqttClient mqttClient;

    /**
     * 發布消息
     * @param topic 主題
     * @param payload 消息內容
     * @param qos QoS級別（0/1/2）
     * @param retained 是否保留消息（true：服務端保留最新消息，新訂閱者可獲取）
     */
    public void publish(String topic, String payload, int qos, boolean retained) throws MqttException {
        if (!mqttClient.isConnected()) {
            mqttClient.reconnect(); // 若連接斷開，先重連
        }
        // 創建MQTT消息
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(qos);
        message.setRetained(retained);
        // 發布消息
        mqttClient.publish(topic, message);
    }

    /**
     * 訂閱主題（重載方法，使用默認QoS）
     */
    public void subscribe(String topic) throws MqttException {
        subscribe(topic, 1); // 默認QoS=1
    }

    /**
     * 訂閱主題
     */
    public void subscribe(String topic, int qos) throws MqttException {
        if (!mqttClient.isConnected()) {
            mqttClient.reconnect();
        }
        mqttClient.subscribe(topic, qos);
        System.out.println("已訂閱主題：" + topic + "（QoS：" + qos + "）");
    }

    /**
     * 取消訂閱主題
     */
    public void unsubscribe(String topic) throws MqttException {
        if (mqttClient.isConnected()) {
            mqttClient.unsubscribe(topic);
            System.out.println("已取消訂閱主題：" + topic);
        }
    }
}

編寫測試接口：驗證 MQTT 功能

@RestController
@RequestMapping("/mqtt")
public class MqttController {

    @Resource
    private MqttUtil mqttUtil;

    /**
     * 發布MQTT消息接口
     * @param topic 主題（如test/topic）
     * @param message 消息內容
     * @param qos QoS級別（0/1/2，默認1）
     * @return 發布結果
     */
    @PostMapping("/publish")
    public String publish(
            @RequestParam String topic,
            @RequestParam String message,
            @RequestParam(required = false, defaultValue = "1") int qos) {
        try {
            // 發布消息（retained=false：不保留消息）
            mqttUtil.publish(topic, message, qos, false);
            return"消息發布成功！主題：" + topic + "，內容：" + message;
        } catch (MqttException e) {
            e.printStackTrace();
            return"消息發布失敗：" + e.getMessage();
        }
    }
}

責任編輯：武曉燕來源：一安未來