Springboot整合mqtt实现软硬件通信的示例代码_Java

前言

本文实现springboot整合mqtt，更好地实现物联网软硬件通信。

一、mqtt是什么？

mqtt 是消息队列遥测传输的缩写，是一种轻量级、基于发布 / 订阅模式的物联网通信协议。

应用场景：

物联网设备通信，比如智能家居、工业传感器、穿戴设备。
远程监控与数据采集，例如环境监测、设备状态上报。

注：这里实现的是软件后端层面的mqtt，是对硬件消息的接收和发送，如果要做一个完整的物联网项目，也需要硬件层面连接mqtt服务器，并且订阅、发布相关主题的信息。

二、物联网项目结构图

该图为完整的mqtt项目结构图。其中mqtt服务器部分可以用官方提供的公共的服务器，也可以在自己的服务器上搭建，只需要在后端配置文件添加mqtt服务器ip地址等相关信息。

注：springboot项目不是自己内部开一个mqtt服务，所以我们只需要连接并使用相应的mqtt服务器即可。

三、实现代码

在pom.xml中引入maven坐标：

        <dependency>
            <groupid>org.springframework.integration</groupid>
            <artifactid>spring-integration-mqtt</artifactid>
        </dependency>

在springboot配置文件中配置mqtt服务器相关信息

spring:
  mqtt:
    url: tcp://broker.emqx.io:1883
    #用户名
    username: admin
    #密码
    password: 123456
    #客户端id（不能重复）
    client:
      id: consumer-id
    #mqtt默认的消息推送主题，实际可在调用接口时指定
    default:
      topic: topic

这里的mqtt服务器地址 tcp://broker.emqx.io:1883 是emqx官方提供的一个公共的服务器，如果是想看一下初步效果，可以使用该服务器。

接下来我们可以放两部分代码，一部分就是接收消息，也就是订阅消息的部分，另一部分就是用来发布相关消息。

订阅消息部分：

@configuration
public class mqttconsumerconfig {
    @value("${spring.mqtt.username}")
    private string username;
    @value("${spring.mqtt.password}")
    private string password;
    @value("${spring.mqtt.url}")
    private string hosturl;
    @value("${spring.mqtt.client.id}")
    private string clientid;
    @value("${spring.mqtt.default.topic}")
    private string defaulttopic;
    /**
     * 客户端对象
     */
    private mqttclient client;
    /**
     * 在bean初始化后连接到服务器
     */
    @postconstruct
    public void init(){
        connect();
    }
    /**
     * 客户端连接服务端
     */
    public void connect(){
        try {
            client = new mqttclient(hosturl,clientid,new memorypersistence());
            mqttconnectoptions options = new mqttconnectoptions();
            options.setcleansession(true);
//            //设置连接用户名
//            options.setusername(username);
//            //设置连接密码
//            options.setpassword(password.tochararray());
            //设置超时时间，单位为秒
            options.setconnectiontimeout(100);
            //设置心跳时间 单位为秒，表示服务器每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
            options.setkeepaliveinterval(20);
            //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
            options.setwill("willtopic",(clientid + "与服务器断开连接").getbytes(),0,false);
            //设置回调
            client.setcallback(new mqttconsumercallback());
            client.connect(options);
            //订阅主题
            //消息等级，和主题数组一一对应，服务端将按照指定等级给订阅了主题的客户端推送消息
            int[] qos = {1};
            system.out.println("连接");
            //主题
            //string[] topics = {"data","status"};
            client.subscribe(topics,qos);
        } catch (mqttexception e) {
            e.printstacktrace();
        }
    }
    /**
     * 断开连接
     */
    public void disconnect(){
        try {
            client.disconnect();
        } catch (mqttexception e) {
            e.printstacktrace();
        }
    }
    /**
     * 订阅主题
     */
    public void subscribe(string topic,int qos){
        try {
            client.subscribe(topic,qos);
        } catch (mqttexception e) {
            e.printstacktrace();
        }
    }
}

@component
public class mqttconsumercallback implements mqttcallback{
    /**
     * 客户端断开连接的回调
     */
    @override
    public void connectionlost(throwable throwable) {
        system.out.println("与服务器断开连接，可重连");
    }
    /**
     * 消息到达的回调
     */
    @override
    public void messagearrived(string topic, mqttmessage message) throws exception {
        system.out.println(string.format("接收消息主题 : %s",topic));
        system.out.println(string.format("接收消息qos : %d",message.getqos()));
        system.out.println(string.format("接收消息内容 : %s",new string(message.getpayload())));
        if(topic.equals("data")){
  system.out.println(string.format("接收消息retained : %b",message.isretained()));
        }
        system.out.println(string.format("接收消息retained : %b",message.isretained()));
    }
    /**
     * 消息发布成功的回调
     */
    @override
    public void deliverycomplete(imqttdeliverytoken imqttdeliverytoken) {
    }
}

接下来是发布消息部分：

@configuration
@slf4j
public class mqttproviderconfig {
    @value("${spring.mqtt.username}")
    private string username;
    @value("${spring.mqtt.password}")
    private string password;
    @value("${spring.mqtt.url}")
    private string hosturl;
    private string clientid = "provider_id";
    @value("${spring.mqtt.default.topic}")
    private string defaulttopic;
    /**
     * 客户端对象
     */
    private mqttclient client;
    /**
     * 在bean初始化后连接到服务器
     */
    @postconstruct
    public void init(){
        connect();
    }
    /**
     * 客户端连接服务端
     */
    public void connect(){
        try{
           // system.out.println("connecting to mqtt server: " + hosturl + " with client id: " + clientid);
            //创建mqtt客户端对象
            client = new mqttclient(hosturl,clientid,new memorypersistence());
            //连接设置
            mqttconnectoptions options = new mqttconnectoptions();
            options.setcleansession(true);
            //设置连接用户名
            options.setusername(username);
            //设置连接密码
            options.setpassword(password.tochararray());
            //设置超时时间，单位为秒
            options.setconnectiontimeout(100);
            //设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线
            options.setkeepaliveinterval(20);
            //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
            options.setwill("willtopic",(clientid + "与服务器断开连接").getbytes(),0,false);
            //设置回调
            client.setcallback(new mqttprovidercallback());
            client.connect(options);
        } catch(mqttexception e){
            e.printstacktrace();
        }
    }
    public void publish(int qos,boolean retained,string topic,string message){
        mqttmessage mqttmessage = new mqttmessage(); //创建消息实例
        mqttmessage.setqos(qos);  //设置qos
        mqttmessage.setretained(retained);  //设置是否保留信息
        mqttmessage.setpayload(message.getbytes());
        //主题的目的地，用于发布/订阅信息
        mqtttopic mqtttopic = client.gettopic(topic);
        //提供一种机制来跟踪消息的传递进度
        //用于在以非阻塞方式（在后台运行）执行发布是跟踪消息的传递进度
        mqttdeliverytoken token;
        try {
            //将指定消息发布到主题，但不等待消息传递完成，返回的token可用于跟踪消息的传递状态
            token = mqtttopic.publish(mqttmessage);
            token.waitforcompletion();
        } catch (mqttexception e) {
            e.printstacktrace();
        }
    }
}

@configuration
public class mqttprovidercallback implements mqttcallback{
    @value("${spring.mqtt.client.id}")
    private string clientid;
    /**
     * 与服务器断开的回调
     */
    @override
    public void connectionlost(throwable cause) {
        system.out.println(clientid+"与服务器断开连接");
    }
    /**
     * 消息到达的回调
     */
    @override
    public void messagearrived(string topic, mqttmessage message) throws exception {
    }
    /**
     * 消息发布成功的回调
     */
    @override
    public void deliverycomplete(imqttdeliverytoken token) {
        imqttasyncclient client = token.getclient();
        system.out.println(client.getclientid()+"发布消息成功！");
    }
}

接下来就是调用publish方法，传入相应参数，即可完成消息的发布。

这里有几个参数要注意一下：

一、mqtt topic（消息路由核心）

topic 是客户端发布 / 订阅消息的 “地址标识”，是实现 “发布 - 订阅” 模式的基础。

核心属性：字符串格式，无预定义结构，由客户端自定义，比如 “device/light/ 客厅”“data/sensor/ 温度”。
层级与通配符：用斜杠 “/” 划分层级，支持两种通配符订阅 ——“+” 匹配单个层级（如 “device/+/ 客厅” 匹配 “device/light/ 客厅”），“#” 匹配当前及所有子层级（如 “data/#” 匹配所有数据类主题）。
核心规则：发布者仅需指定 topic 发送消息，订阅者通过匹配 topic（或通配符）接收消息，发布者与订阅者无直接关联，实现解耦。

二、mqtt qos（消息传输质量等级）

qos（quality of service）定义了消息从发布者到订阅者的传输可靠性，mqtt 3.1.1 标准规定了 3 个等级，优先级从低到高。