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

在spring boot项目中集成mqtt协议，可以方便地实现与物联网设备或其他支持mqtt的系统进行通信。以下是详细的步骤和代码示例，帮助您快速完成spring boot与mqtt的集成。

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</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相关参数，例如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. 创建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 connectionlost(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) {
            return "订阅失败：" + e.getmessage();
        }
    }
}

7. 启动并测试

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