一篇文章搞透MQTT的基本使用以及与SpringBoot的整合

一、MQTT协议介绍

MQTT(Message Queuing Telemetry Transport，消息队列遥测传输协议)是一种基于发布/订阅(publish/subscribe)模式的轻量级通讯协议，基于TCP/IP协议构建。MQTT的最大优点在于可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网小型设备、移动应用等方面有较广泛的应用。

MQTT特点：

1）使用发布/订阅消息模式，提供一对多的消息发布，解除应用的耦合。
2）二进制形式编码，小型化传输，降低网络开销。
3）三种消息发布服务质量保证(QOS)：
	·qos=0：“至多一次”，可能会发生消息丢失或重复，消息发布依赖于TCP/IP网络。
	·qos=1：“至少一次”，确保消息到达，但消息可能会重复。
	·qos=2：“只有一次”，确保消息只会到达一次。
4）采用心跳机制，间断性发送报文维持客户端和服务端的长连接，减少电量的消耗，提升系统资源利用率。

MQTT中也有主题的概念，用于隔离各种繁杂的消息。不同的主题可被不同的订阅者订阅，发布者可以发布不同的主题消息，从而将消息进行隔离区分。MQTT协议规定主题是UTF-8编码的字符串。主题的命名规则：

1）所有的主题名必须至少包含一个字符。
2）主题名大小写敏感，例如ACCOUNTS和Accounts是不同的主题名。
3）主题名可以包含空格字符，例如Accounts payable是合法的主题名。
4）主题名以前置或后置斜杠/区分，例如/finance和finance是不同的。
5）只包含斜杠/的主题名是合法的。
6）主题名不能包含null字符(Unicode U+0000)。

MQTT主题通配符包含单层通配符“+”及多层通配符“#”，主要用于客户端一次订阅多个主题。注意，通配符只能用于订阅，不能用于发布。

1）单层通配符：如客户端订阅AAA/+/CCC主题消息，则会收到AAA/1/CCC、AAA/2/CCC、…、AAA/N/CCC主题消息。
2）多层通配符：如客户端订阅AAA/#主题消息，则会收到AAA、AAA/BBB、AAA/BBB/CCC主题消息。
3）以$开头的主题为系统主题。

二、MQTT的基本使用

2.1 MQTT服务器EMQX搭建

2.1.1 官方安装手册及文档

EMQX是一款全球下载量超千万的开源物联网MQTT服务器，单集群支持1亿物联网设备连接，消息分发时延低于1毫秒，助力企业构建关键业务的IOT平台与应用。下载和参考地址：https://www.emqx.io/zh/downloads

使用手册：

2.1.2 Linux环境下使用Docker部署EMQX

使用Docker安装快速体验(更多安装方式参考上一节)：

docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:latest

常见端口：

1883：MQTT TCP协议端口
8883：MQTT/TCP SSL端口
8083：MQTT/WebSocket端口
8084：MQTT/WebSocket with SSL端口
8080：MQTT执行引擎HTTP API端口
18083：EMQX Dashboard管理控制台端口

浏览器访问控制台页面，默认账号密码为admin/public，地址：http://IP:18083

2.1.3 Windows环境下安装EMQX

下载安装包：

解压安装包到指定目录：

命令行下进入bin目录，启动EMQX：

emqx start

浏览器访问控制台页面，默认账号密码为admin/public，地址：http://127.0.0.1:18083/

2.1.4 EMQX配置文件介绍

EMQX主配置文件为 emqx.conf，根据安装方式其所在位置有所不同：

参考地址：https://www.emqx.io/docs/zh/latest/configuration/configuration.html

相关配置：

# 当前节点配置
node {
  name = "emqx@127.0.0.1"
  cookie = "emqxsecretcookie"
  data_dir = "/var/lib/emqx"
}

# 日志配置
log {
  file_handlers.default {
    level = warning
    file = "/var/log/emqx/emqx.log"
  }
}

# 集群配置
cluster {
  name = emqxcl
  discovery_strategy = manual
}

# MQTT TCP连接端口号
listeners.tcp.default {
  bind = "0.0.0.0:1883"
  max_connections = 1024000
}

# MQTT HTTPS连接端口号
listeners.ssl.default {
  bind = "0.0.0.0:8883"
  max_connections = 512000
  ssl_options {
    keyfile = "/etc/emqx/certs/key.pem"
    certfile = "/etc/emqx/certs/cert.pem"
    cacertfile = "/etc/emqx/certs/cacert.pem"
  }
}

# MQTT WebSocket端口
listeners.ws.default {
  bind = "0.0.0.0:8083"
  max_connections = 1024000
  websocket.mqtt_path = "/mqtt"
}

# MQTT WebSocket with SSL端口
listeners.wss.default {
  bind = "0.0.0.0:8084"
  max_connections = 512000
  websocket.mqtt_path = "/mqtt"
  ssl_options {
    keyfile = "/etc/emqx/certs/key.pem"
    certfile = "/etc/emqx/certs/cert.pem"
    cacertfile = "/etc/emqx/certs/cacert.pem"
  }
}

# 面板设置，例如网页访问的端口
dashboard {
    listeners.http {
        bind = 18083
    }
    default_username = "admin"
    default_password = "public"
}

# 权限相关设置
authorization {
  deny_action = ignore
  no_match = allow
  cache = { enable = true }
  sources =  [
    {
      type = file
      enable = true
      # This file is immutable to EMQX.
      # Once new rules are created from dashboard UI or HTTP API,
      # the file 'data/authz/acl.conf' is used instead of this one
      path = "/etc/emqx/acl.conf"
    }
  ]
}

2.2 EMQX开机自启动问题

2.2.1 Linux系统

以Centos系统为例，安装教程在官方文档中已经提到：https://www.emqx.io/zh/downloads

创建开机自启动脚本：

[root@localhost local]# vi /etc/init.d/emqx

#!/bin/sh
#
# emqttd       
#
# Startup script for emqttd.
#
# chkconfig: 2345 90 10
# description: emqttd is mqtt broker.
#
# source function library
. /etc/rc.d/init.d/functions
 
export HOME=/root

start() {
    echo "starting emqttd..."
    # 此处根据实际安装目录修改下面的路径
    cd /home/emqx && ./bin/emqx start
}

stop() {
    echo "stopping emqttd..."
    # 此处根据实际安装目录修改下面的路径
    cd /home/emqx && ./bin/emqx stop
}

restart() {
    stop
    start
}

case "$1" in
    start)
        start
        ;;
    stop)
        stop
        ;;
    restart)
        restart
        ;;
    *)
        echo $"Usage: $0 {start|stop}"
        RETVAL=2
esac

保存退出，然后运行以下命令将脚本注册为服务：

[root@localhost local]# chmod +x /etc/init.d/emqx
[root@localhost local]# chkconfig --add emqx

启动和停止相关指令：

启动服务：systemctl start emqx
停止服务：systemctl stop emqx
重启服务：systemctl restart emqx

2.2.2 Windows系统

1.最简单的方式

在EMQX解压bin目录下创建start.bat脚本(脚本名称任意)：、

@echo off
cmd /c "emqx start"
exit

同级目录，新建start.vbs文件，写入如下命令(命令最后一个0是隐藏窗口运行)：

createobject("wscript.shell").run"D:\dmp\start.bat",0

然后右键创建start.vbs的快捷方式，移动到开机自启动目录下(也可以使用win+r快捷键，输入shell:startup命令进入)：

C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Startup

这样就可以重启测试了。

2.定时判断

创建bat文件：

@echo off
%1 mshta vbscript:CreateObject("WScript.Shell").Run("%~s0 ::",0,FALSE)(window.close)&&exit

set EMQX_PATH=D:\emqx-5.1.4
set OUTPUT_FILE=%EMQX_PATH%\emqx_monitor.log

:loop
cd /d "%EMQX_PATH%"

REM 获取当前日期时间，并格式化为时间戳
for /f "tokens=1-4 delims=/ " %%a in ('date /t') do (
    set year=%%a
    set month=%%b
    set day=%%c
)
for /f "tokens=1-3 delims=: " %%a in ('time /t') do (
    set hour=0%%a
    set minute=%%b
)
set timestamp=%year%-%month%-%day% %hour:~-2%:%minute%

cd bin
emqx_ctl status | findstr /C:"Node 'emqx@127.0.0.1'" | findstr /C:"is started" >NUL
IF "%ERRORLEVEL%"=="0" (
    echo [%timestamp%] EMQX is running normally. >> %OUTPUT_FILE%
) ELSE (
    echo [%timestamp%] EMQX is not running. Restarting... >> %OUTPUT_FILE%

    REM 结束之前的 EMQX 进程，可以通过 taskkill 命令来实现
    REM taskkill /f /im beam.smp.exe >NUL

    start /b cmd /c "%EMQX_PATH%\bin\emqx start"

    echo [%timestamp%] EMQX has started. >> %OUTPUT_FILE%

    REM 如果 EMQX 不直接启动一个新窗口，可以将上面一行修改为以下内容
    REM call %EMQX_PATH%\emqx start
)

timeout /t 60 > NUL
goto loop

这个批处理文件会不断检查EMQX服务的状态，并根据服务的运行状态记录相关信息到日志文件中。如果服务未运行，它将尝试重新启动服务。然后，它会等待一段时间后再次进行状态检查，以持续监视服务的运行情况。注意EMQX_PATH换成自己实际的路径，然后鼠标右键创建快捷方式，将快捷方式移动到开机自启动目录下(也可以使用win+r快捷键，输入shell:startup命令进入)：

C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Startup

日志文件位于emqx目录下的 emqx_monitor.log，如下所示：

2.3 MQTT协议客户端工具MQTTX

2.3.1 MQTTX安装

MQTTX是一个强大的跨平台MQTT 5.0桌面和CLI客户端，使学习、探索和开发 MQTT 变得快速而简单。下载地址：https://mqttx.app/zh#download

下载后点击安装：

打开后主界面如下：

MQTTX的使用手册(建议仔细阅读)：https://mqttx.app/zh/docs

2.3.2 MQTTX及EMQX的测试与使用

2.3.2.1 快速建立连接

使用MQTTX测试(默认EMQX未开启用户认证)：

查看服务端网页：

2.3.2.2 消息的发布与订阅

1.主界面

连接成功后，即可进入连接的主界面。点击顶部连接名称旁的折叠按钮，可以查看并修改该连接的基本信息。在断开连接的情况下，修改这些常用配置项需要重新连接后才能生效。每次连接成功后，面板都会自动折叠。断开连接时，您也可以点击右边的配置按钮进行更多连接配置的修改。

2.添加订阅

点击左下角的 New Subscription 按钮，可以快速添加一个 Topic。每个 Topic 可以用一种颜色进行标记，颜色可以随机生成或通过颜色选择器进行选择。添加完成后，点击订阅列表中的已订阅Topic，可以实现消息过滤。此时消息视图中只会显示当前选中主题的消息内容，再次点击即可取消过滤。

在添加Topic时，可以给每个Topic设置一个别名(别名是可选项)。设置别名并完成订阅后，订阅列表中的Topic将以别名显示。鼠标悬浮在Topic项上时，提示框将同时显示出该Topic的原值。这在需要监控查看的多个Topic过长，难以辨别Topic具体含义时，有很大帮助。

在Topic输入框中输入多个Topic，使用逗号(,)进行分割，点击确认订阅成功后，订阅列表中会包含多个Topic。对于使用客户端层别名功能，也可以同时对多个Topic进行设置，同样使用逗号分隔(,)。

订阅的主题支持编辑/禁用/启用。右键点击已经订阅过的Topic列表项，在右键菜单中我们可以快速选择编辑、禁用或启用等操作。当订阅的Topic过多时，如果并非所有的Topic消息都需要接收，您可以使用禁用功能避免再次订阅相同的Topic。需要重新接收该Topic的消息时，只需再次启用即可。

3.消息发送与接收

Topic订阅成功后，就可以测试消息的收发了。在页面右下角填入选中订阅的Topic信息，可选择QoS和Retain的值，输入payload(载荷/消息体)后，点击最右侧的发送按钮，就向刚才订阅的Topic发送了一条消息。发送成功后，也可以立即接收到刚才所发送的消息。注意，消息框内，右侧是发送的消息，左侧是接收到的消息。对于MacOS用户，可以使用command + enter快捷键，其它用户可以使用control + enter快捷键来快速发送消息。消息栏的payload选项，支持将消息快速转化为Base64、Hex、JSON和Plaintext四种格式。

至此，就完成了一条MQTT消息的测试。

三、SpringBoot实现发布和订阅MQTT消息

3.1 SpringBoot整合MQTT

引入依赖：

<dependency>
	<groupId>org.eclipse.paho</groupId>
	<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
	<version>1.2.5</version>
</dependency>

项目的配置文件：

server:
  port: 9001

spring:
  main:
    allow-circular-references: true # 允许循环依赖

# MQTT配置
mqtt:
  url: "tcp://localhost:1883" # mqtt连接地址+端口
  username: admin # mqtt连接账号(这里使用默认)
  password: public # mqtt连接密码(这里使用默认)
  client_id: mqtt_client_xiange # ClientID，可以自定义，在EMQX服务端可以看到连接名
  default_topic: character # 默认连接的主题，多个可以使用逗号,分割
  timeout: 100 # 连接超时时间
  keepalive: 60 # 心跳间隔

新建配置类，此类在getMqttPushClient()方法中会发起连接和加载application.yml的配置：

import com.yuqixian.mqtt_client.client.MqttPushClient;
import com.yuqixian.mqtt_client.client.MqttSubClient;
import lombok.Data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Component
@Data
@ConfigurationProperties(value = "mqtt")
public final class MqttConfig {

    @Autowired
    private MqttPushClient mqttPushClient;

    @Autowired
    private MqttSubClient mqttSubClient;

    /**
     * MQTT连接地址
     */
    private String url;

    /**
     * MQTT用户名
     */
    private String username;

    /**
     * MQTT密码
     */
    private String password;

    /**
     * ClientId(客户端ID)
     */
    private String client_id;

    /**
     * 默认连接主题
     */
    private String default_topic;

    /**
     * 超时时间
     */
    private int timeout;

    /**
     * 保持连接数
     */
    private int keepalive;

    /**
     * 连接至mqtt服务器，获取mqtt连接
     *
     * @return
     */
    @Bean
    public MqttPushClient getMqttPushClient() {
        // 连接至mqtt服务器，获取mqtt连接
        mqttPushClient.connect(url, client_id, username, password, timeout, keepalive);
        // 一连接mqtt，就订阅默认需要订阅的主题
        mqttSubClient.subscribDefaultTopic(default_topic);
        return mqttPushClient;
    }

}

 创建发布连接类，此类中包含连接方法以及发布方法，因MQTT服务有可能中断，所以在连接时需要设置自动重连，并且在重连时要重新订阅，此过程在回调中实现：

import com.yuqixian.mqtt_client.callback.PushCallback;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 发布连接类
 */
@Slf4j
@Component
public class MqttPushClient {
    @Autowired
    private PushCallback pushCallback;

    private static MqttClient client;

    public static void setClient(MqttClient client) {
        MqttPushClient.client = client;
    }

    public static MqttClient getClient() {
        return client;
    }

    public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) {
        MqttClient client;
        try {
            client = new MqttClient(host, clientID, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(timeout);
            options.setKeepAliveInterval(keepalive);
            // mqtt服务有可能中断，automaticReconnect为true表示断线自动重连，但仅仅只是重新连接，并不订阅主题；在connectComplete回调函数重新订阅
            options.setAutomaticReconnect(true);
            MqttPushClient.setClient(client);
            try {
                // 设置回调类
                client.setCallback(pushCallback);
                // client.connect(options);
                IMqttToken iMqttToken = client.connectWithResult(options);
                boolean complete = iMqttToken.isComplete();
                log.info("MQTT连接" + (complete ? "成功" : "失败"));
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 发布，默认qos为0，非持久化
     *
     * @param topic       主题名
     * @param pushMessage 消息
     */
    public void publish(String topic, String pushMessage) {
        publish(0, false, topic, pushMessage);
    }

    /**
     * 发布
     *
     * @param qos
     * @param retained
     * @param topic
     * @param pushMessage
     */
    public void publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic);
        if (null == mTopic) {
            log.error("主题不存在:{}", mTopic);
        }
        try {
            mTopic.publish(message);
        } catch (Exception e) {
            log.error("mqtt发送消息异常:", e);
        }
    }

}

创建订阅类，该类包含订阅配置的主题方法以及普通订阅方法：

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.stereotype.Component;

/**
 * 订阅类
 */
@Slf4j
@Component
public class MqttSubClient {

    public void subscribDefaultTopic(String default_topic) {
        if(!StringUtils.isEmpty(default_topic)) {
            // 订阅test_queue主题
            String mqtt_topic[] = default_topic.split(",");
            if(mqtt_topic != null && mqtt_topic.length > 0) {
                for (int i = 0; i < mqtt_topic.length; i++) {
                    subscribe(mqtt_topic[i], 0);// 订阅主题
                }
            }
        }
    }

    /**
     * 订阅某个主题，qos默认为0
     *
     * @param topic
     */
    public void subscribe(String topic) {
        subscribe(topic, 0);
    }

    /**
     * 订阅某个主题
     *
     * @param topic 主题名
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            MqttClient client = MqttPushClient.getClient();
            if (client == null) return;
            client.subscribe(topic, qos);
            log.info("订阅主题：{}", topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

}

创建回调类，此类包含断开重连，所有被订阅的主题消息都将在这里被接收：

import com.yuqixian.mqtt_client.config.MqttConfig;
import com.yuqixian.mqtt_client.factory.MqttFactory;
import com.yuqixian.mqtt_client.service.MqttService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 回调类
 */
@Slf4j
@Component
public class PushCallback implements MqttCallback {
    @Autowired
    private MqttConfig mqttConfig;
    @Override
    public void connectionLost(Throwable cause) {
        // 连接丢失后，一般在这里面进行重连
        log.info("连接断开，正在重连......");
        mqttConfig.getMqttPushClient();
    }

    /**
     * 发送消息，消息到达后处理方法
     * @param token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // log.info("deliveryComplete ------" + token.isComplete());
    }

    /**
     * 订阅主题接收到消息处理方法
     * @param topic
     * @param message
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) {
        // subscribe后得到的消息会执行到这里面,这里在控制台有输出
        log.info("接收消息主题：" + topic);
        log.info("接收消息Qos：" + message.getQos());
        log.info("接收消息内容：" + new String(message.getPayload()));

        MqttService mqttService = MqttFactory.getMqttFactory(topic);
        if (mqttService != null){
            mqttService.analysisMessage(new String(message.getPayload()));
        } else {
            log.error("接收topic异常：{}",topic);
        }
    }

}

消息的具体处理选择工厂设计模式，方便后续处理：

import com.yuqixian.mqtt_client.service.MqttService;
import com.yuqixian.mqtt_client.service.impl.CharacterMqttServiceImpl;
import com.yuqixian.mqtt_client.service.impl.HelloMqttServiceImpl;

public class MqttFactory {

    public static MqttService getMqttFactory(String topic){
        switch (topic){
            case "character":
                return new CharacterMqttServiceImpl();
            case "hello":
                return new HelloMqttServiceImpl();
            default :
                return null;
        }
    }
}

对应的接口和实现类：

public interface MqttService {

    void analysisMessage(String message);

}


import com.yuqixian.mqtt_client.service.MqttService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class HelloMqttServiceImpl implements MqttService {

    @Override
    public void analysisMessage(String message) {
        log.info("处理topic/hello信息：{}", message);
    }

}


import com.yuqixian.mqtt_client.service.MqttService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class CharacterMqttServiceImpl  implements MqttService {

    @Override
    public void analysisMessage(String message) {
        log.info("处理topic/character信息：{}", message);
    }

}

添加Controller方便测试：

import com.yuqixian.mqtt_client.client.MqttPushClient;
import com.yuqixian.mqtt_client.client.MqttSubClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
@Slf4j
public class MqttClientController {

    @Autowired
    MqttPushClient mqttPushClient;

    @Autowired
    MqttSubClient mqttSubClient;

    @GetMapping("/push")
    public String push(@RequestParam("message") String message){
        log.info("调用");
        mqttPushClient.publish("character", message);
        return "发送成功";
    }

    @GetMapping("/subscribe")
    public String subscribe(String topic){
        mqttSubClient.subscribe(topic);
        return "接收成功";
    }

}

3.2 测试

前面已经把MQTTX及EMQX部署好了，先让MQTTX订阅character的主题：

测试消息正常收发：

启动项目，控制台显示连接成功：

查看EMQX管控台，多了一个连接的客户端：

使用MQTTX向指定主题发送一条消息：

查看控制台日志，程序收到消息：

接下来测试发送消息：http://127.0.0.1:9001/push?message=tryagain

查看控制台，消息发送成功，并且可以收到订阅的消息：

另一个客户端MQTTX也可以收到消息：

博文参考：

MQTT入门基础概念与学习介绍及服务部署搭建并使用桌面工具进行测试开发：https://blog.csdn.net/wfeil211/article/details/128925822
如何使用SpringBoot项目对接mqtt实现发布者与订阅者进行消息处理：https://blog.csdn.net/wfeil211/article/details/128988229
Java实现MQTT协议：https://blog.csdn.net/weixin_44905972/article/details/131665070
Windows开机自启动EMQX：https://blog.csdn.net/weixin_44285713/article/details/128779697
https://blog.csdn.net/qq_32024669/article/details/131536312
Linux开机自启动EMQX：https://blog.csdn.net/qq_40083897/article/details/128417331