快捷導(dǎo)航

如何在Spring Boot項(xiàng)目中集成MQTT協(xié)議

更新時(shí)間：2025年06月24日 14:36:30 作者：getapi

本文介紹在SpringBoot中集成MQTT的步驟,包括安裝Broker、添加Eclipse Paho依賴、配置連接參數(shù)、實(shí)現(xiàn)消息發(fā)布訂閱、測試接口及注意事項(xiàng),如唯一客戶端ID、安全措施等,幫助開發(fā)者快速實(shí)現(xiàn)物聯(lián)網(wǎng)通信,感興趣的朋友一起看看吧

1. 準(zhǔn)備工作

在開始之前，請確保以下條件已滿足：

已安裝并運(yùn)行MQTT Broker（如 Eclipse Mosquitto 或 EMQX）。
Spring Boot項(xiàng)目已創(chuàng)建（可以通過 Spring Initializr 快速生成）。

2. 引入依賴

在pom.xml文件中添加必要的依賴項(xiàng)。我們使用 Eclipse Paho MQTT Client 作為MQTT客戶端庫。

<dependencies>
    <!-- Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <!-- Eclipse Paho MQTT Client -->
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.5</version>
    </dependency>
</dependencies>

3. 配置MQTT連接

在application.yml或application.properties中配置MQTT相關(guān)參數(shù)，例如Broker地址、客戶端ID等。

application.yml 示例：

mqtt:
  broker-url: tcp://localhost:1883
  client-id: springboot-mqtt-client
  username: mqtt_user
  password: mqtt_password
  topic: test/topic
  qos: 1

4. 創(chuàng)建MQTT配置類

創(chuàng)建一個(gè)配置類來初始化MQTT客戶端，并設(shè)置連接選項(xiàng)、回調(diào)函數(shù)等。

import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MqttConfig {
    @Value("${mqtt.broker-url}")
    private String brokerUrl;
    @Value("${mqtt.client-id}")
    private String clientId;
    @Value("${mqtt.username}")
    private String username;
    @Value("${mqtt.password}")
    private String password;
    @Bean
    public MqttClient mqttClient() throws MqttException {
        MqttClient client = new MqttClient(brokerUrl, clientId, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(10);
        options.setKeepAliveInterval(20);
        // 設(shè)置回調(diào)
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("MQTT連接丟失：" + cause.getMessage());
            }
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("收到消息 - 主題：" + topic + "，內(nèi)容：" + new String(message.getPayload()));
            }
            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("消息發(fā)送完成");
            }
        });
        // 連接到MQTT Broker
        client.connect(options);
        return client;
    }
}

5. 實(shí)現(xiàn)消息發(fā)布與訂閱

創(chuàng)建服務(wù)類來處理消息的發(fā)布和訂閱操作。

消息發(fā)布服務(wù)

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class MqttPublishService {
    @Autowired
    private MqttClient mqttClient;
    @Value("${mqtt.topic}")
    private String topic;
    @Value("${mqtt.qos}")
    private int qos;
    public void publish(String payload) throws Exception {
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(qos);
        mqttClient.publish(topic, message);
        System.out.println("消息已發(fā)布 - 主題：" + topic + "，內(nèi)容：" + payload);
    }
}

消息訂閱服務(wù)

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class MqttSubscribeService {
    @Autowired
    private MqttClient mqttClient;
    @Value("${mqtt.topic}")
    private String topic;
    @Value("${mqtt.qos}")
    private int qos;
    public void subscribe() throws Exception {
        mqttClient.subscribe(topic, qos);
        System.out.println("已訂閱主題：" + topic);
    }
}

6. 測試功能

在控制器中調(diào)用發(fā)布和訂閱服務(wù)，測試MQTT功能。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
@RestController
@RequestMapping("/mqtt")
public class MqttController {
    @Autowired
    private MqttPublishService mqttPublishService;
    @Autowired
    private MqttSubscribeService mqttSubscribeService;
    @PostMapping("/publish")
    public String publishMessage(@RequestParam String message) {
        try {
            mqttPublishService.publish(message);
            return "消息發(fā)布成功";
        } catch (Exception e) {
            return "消息發(fā)布失?。? + e.getMessage();
        }
    }
    @GetMapping("/subscribe")
    public String subscribeTopic() {
        try {
            mqttSubscribeService.subscribe();
            return "訂閱成功";
        } catch (Exception e) {
            return "訂閱失?。? + e.getMessage();
        }
    }
}

7. 啟動(dòng)并測試

啟動(dòng)Spring Boot應(yīng)用程序。
使用Postman或?yàn)g覽器訪問以下接口：
- 發(fā)布消息：POST http://localhost:8080/mqtt/publish?message=HelloMQTT
- 訂閱主題：GET http://localhost:8080/mqtt/subscribe
檢查控制臺(tái)輸出，驗(yàn)證消息是否正確發(fā)布和接收。