如何在Spring Boot项目中集成MQTT协议

本文主要是介绍如何在Spring Boot项目中集成MQTT协议，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

《如何在SpringBoot项目中集成MQTT协议》本文介绍在SpringBoot中集成MQTT的步骤,包括安装Broker、添加EclipsePaho依赖、配置连接参数、实现消息发布订阅、测试接口...

1. 准备工作

在开始之前，请确保以下条件已满足：

已安装并运行MQTT Broker（如 Eclipse Mosquitto 或 EMQX）。
Spring Boot项目已创建（可以通过 Spring Initializr 快速生成）。

2. 引入依赖

在pom.XML文件中添加必要的依赖项。我们使用 Eclipse Paho MQTT Client 作为MQTT客户端库。

<dependencies>
    <!-- Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artvkuJKjAifactId>
    </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相关参数，例如Broker地址、客户端ID等。

application.yml 示例：

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

4. 创建MQTT配置类

创建一个配置类来初始化MQTT客户端，并设置连接选项、回调函数等。

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);
        // 设置回调
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLosjst(Throwable cause) {
                System.out.println("MQTT连接丢失：" + cause.getMessage());
            }
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("收到消息 - 主题：" + topic + "，内容：" + new String(message.getPayload()));
            }
            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("消息发送完成");
            }
        });
        // 连接到MQTT Broker
        client.connect(options);
        return client;
    }
}

5. 实现消息发布与订阅

创建服务类来处理消息的发布和订阅操作。

消息发布服务

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("消息已发布 - 主题：" + topic + "，内容：" + payload);
    }
}

消息订阅服务

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. 测试功能

在控制器中调用发布和订阅服务，测试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 "消息发布成功";
        } catch (Exception e) {
            return "消息发布失败：" + e.getMessage();
        }
    }
    @GetMapping("/subscribe")
    public String subscribeTopic() {
        try {
            mqttSubscribeService.subscribe();
            return "订阅成功";
        } catch (Exception e) {
           http://www.chinasem.cn return "订阅失败：" + e.getMessage();
        }
    }
}

7. 启动并测试

启动Spring Boot应用程序。
使用Postman或浏览器访问以下接口：
- 发布消息：POST http://localhost:8080/mqtt/publish?message=HelloMQTT
- 订阅主题：GET http://localhost:8080/mqtt/subscribe
检查控制台输出，验证消息是否正确发布和接收。