使用Springboot实现MQTT通信

news2025/2/8 6:03:16

目录

一、MQ协议

MQTT 特点

MQTT 工作原理

MQTT 主要应用场景

MQTT 配置与注意事项

二、MQTT服务器搭建

三、参考案例

MQTT（Message Queuing Telemetry Transport）是一种基于发布/订阅模型的轻量级消息传输协议，常用于物联网（IoT）场景中。它设计简洁、带宽占用少，非常适合资源受限的设备和网络环境。

一、MQ协议

MQTT 特点

  1. 轻量级协议：

    • 设计简单，占用带宽少，特别适合嵌入式设备和不稳定的网络环境。

  2. 发布/订阅模型：

    • 客户端通过主题（Topic）发布消息，订阅者通过主题接收消息，彼此不直接通信。

  3. 可靠性保障：

    • 提供三种服务质量（QoS）等级，确保消息可靠传输：
      • QoS 0：至多一次（不确认，可能丢失）。
      • QoS 1：至少一次（需要确认，但可能重复）。
      • QoS 2：仅一次（确保消息不丢失且不重复）。

  4. 持续连接：

    • 使用 TCP/IP 连接，通过心跳包（Keep-Alive）保持连接稳定。

  5. 支持离线消息：

    • 使用“保留消息”和“持久会话”功能，实现离线设备接收消息。

  6. 安全性：

    • 支持 SSL/TLS 加密，结合用户名和密码进行身份验证。
MQTT 工作原理

  1. 连接：

    • 客户端通过 CONNECT 消息向服务器建立连接，服务器返回 CONNACK 消息。

  2. 发布：

    • 客户端通过 PUBLISH 消息向服务器发布消息，指定消息的主题。

  3. 订阅：

    • 客户端通过 SUBSCRIBE 消息订阅一个或多个主题，服务器将匹配主题的消息推送给客户端。

  4. 心跳：

    • 客户端和服务器定期发送心跳包（PINGREQ 和 PINGRESP），确保连接有效。

  5. 断开：

    • 客户端通过 DISCONNECT 消息通知服务器主动断开连接。
MQTT 主要应用场景

  1. 物联网（IoT）：

    • 设备状态监控、数据收集和远程控制。

  2. 智能家居：

    • 控制家电、监控传感器数据。

  3. 车联网：

    • 实时车辆数据传输、位置追踪。

  4. 移动应用：

    • 消息推送、实时聊天。

  5. 工业领域：

    • 设备数据采集和分析。
MQTT 配置与注意事项

  1. 主题命名：

    • 使用层级结构（如 /iot/device/status），便于管理。
    • 避免过于复杂的主题结构。

  2. QoS 选择：

    • 根据应用需求选择适合的 QoS 等级，平衡可靠性和性能。

  3. 安全措施：

    • 启用 SSL/TLS 加密。
    • 配置用户名和密码，限制匿名连接。
    • 控制主题的访问权限。

  4. 性能优化：

    • 控制消息大小，减少带宽占用。
    • 调整心跳时间，优化连接稳定性。

二、MQTT服务器搭建

1、在springboot项目工程pom文件下引入相关依赖

        <!--mqtt相关依赖-->
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-stream</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
        </dependency>

2、修改application.yml配置文件

spring:
  application:
    name: provider
    #MQTT配置信息
  mqtt:
     #MQTT服务地址，端口号默认1883，如果有多个，用逗号隔开
    url: tcp://127.0.0.1:1883
     #用户名
    username: guest
     #密码
    password: guest
     #客户端id(不能重复)
    client:
        id: provider-id
    #MQTT默认的消息推送主题，实际可在调用接口是指定
    default: 
        topic: topic
server:
  port: 8080

3、消息发布者客户端配置

?
package com.three.demo.mqtt.config;

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.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;

@Configuration
@Slf4j
public class MqttClientConfig {

    @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;

    @Autowired
    private MqttClientCallBack mqttClientCallBack;

    /**
     * 客户端对象
     */
    private MqttAsyncClient client;

    /**
     * 在bean初始化后连接到服务器
     */
    @PostConstruct
    public void init() {
        connect();
    }

    /**
     * 客户端连接服务端
     */
    public void connect() {
        //连接设置
        MqttConnectOptions options = new MqttConnectOptions();
        //是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos）,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
        //设置为true表示每次连接服务器都是以新的身份
        options.setCleanSession(false);
        //设置连接用户名
        options.setUserName(username);
        //设置连接密码
        options.setPassword(password.toCharArray());
        //设置超时时间，单位为秒
        options.setConnectionTimeout(60);
        //设置心跳时间 单位为秒，表示服务器每隔 1.5*10秒的时间向客户端发送心跳判断客户端是否在线
        options.setKeepAliveInterval(20);
        // 开启自动重连
        options.setAutomaticReconnect(true);
        // 设置最大重连时间间隔 (可选)，单位是毫秒，设置为 5000 表示最多等待 5 秒再尝试重连
        options.setMaxReconnectDelay(5000);
        //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
        options.setWill("willTopic", (clientId + "与服务器断开连接").getBytes(), 0, false);
        try {
            //创建MQTT客户端对象
            client = new MqttAsyncClient(hostUrl, clientId, new MemoryPersistence());
            //设置回调
            client.setCallback(mqttClientCallBack);
            // 使用异步连接
            client.connect(options, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    log.info("MQTT连接成功");
                }
                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    log.error("MQTT连接失败：" + exception.getMessage());
                }
            });
        } catch (MqttException e) {
            log.error("mqtt连接失败。。" + e.getMessage());
        }
    }

    public void publish(int qos, boolean retained) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);
        mqttMessage.setRetained(retained);
        mqttMessage.setPayload(pushLog.getData().getBytes());
        try {
            // 使用异步客户端发布消息，并处理结果
            client.publish(pushLog.getTopic(), mqttMessage, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    System.out.println("发送成功");
                }
                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    log.error("发送失败：" + exception.getMessage());
                }
            });
        } catch (MqttException e) {
            log.error("发送失败：" + e.getMessage());
        }
    }

    /**
     * 断开连接
     */
    public void disConnect() {
        try {
            client.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

?

4、消息发布客户端回调

package com.three.demo.mqtt.config;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class MqttClientCallBack implements MqttCallback {

    @Value("${spring.mqtt.client.id}")
    private String clientId;

    /**
     * 与服务器断开的回调
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.error(clientId + "与服务器断开连接！！" + cause.getMessage());
    }

	/**
     * 消息发布成功的回调
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        IMqttAsyncClient client = token.getClient();
        System.out.println(client.getClientId()+"发布消息成功！");
        
    }

}

5、创建控制器测试发布信息

package com.three.demo.mqtt.controller;
 
import com.three.demo.mqtt.config.MqttClientConfig;
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
 
@Controller
public class SendController {
    @Autowired
    private MqttClientConfig client;
 
    @RequestMapping("/sendMessage")
    @ResponseBody
    public String sendMessage(int qos,boolean retained,String topic,String message){
        try {
            client.publish(qos, retained, topic, message);
            return "发送成功";
        } catch (Exception e) {
            e.printStackTrace();
            return "发送失败";
        }
    }
}

6、消息接收者配置

这里我对之前的代码进行改造

/**
     * 客户端连接服务端
     */
    public void connect() {
        //连接设置
        MqttConnectOptions options = new MqttConnectOptions();
        //是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos）,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
        //设置为true表示每次连接服务器都是以新的身份
        options.setCleanSession(false);
        //设置连接用户名
        options.setUserName(username);
        //设置连接密码
        options.setPassword(password.toCharArray());
        //设置超时时间，单位为秒
        options.setConnectionTimeout(60);
        //设置心跳时间 单位为秒，表示服务器每隔 1.5*10秒的时间向客户端发送心跳判断客户端是否在线
        options.setKeepAliveInterval(20);
        // 开启自动重连
        options.setAutomaticReconnect(true);
        // 设置最大重连时间间隔 (可选)，单位是毫秒，设置为 5000 表示最多等待 5 秒再尝试重连
        options.setMaxReconnectDelay(5000);
        //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
        options.setWill("willTopic", (clientId + "与服务器断开连接").getBytes(), 0, false);
        try {
            //创建MQTT客户端对象
            client = new MqttAsyncClient(hostUrl, clientId, new MemoryPersistence());
            //设置回调
            client.setCallback(mqttClientCallBack);
            // 使用异步连接
            client.connect(options, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    log.info("MQTT连接成功");
                    // 连接成功后订阅主题
                    try {
                        //订阅主题
                        //消息等级，和主题数组一一对应，服务端将按照指定等级给订阅了主题的客户端推送消息
                        int[] qos = {2, 2};
                        String[] topics = {
                                "/iot/msg/topic1",
                                "/iot/msg/topic2"
                        };
                        client.subscribe(topics, qos);
                        log.info("订阅主题成功");
                    } catch (MqttException e) {
                        log.error("订阅主题失败：" + e.getMessage());
                    }
                }
                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    log.error("MQTT连接失败：" + exception.getMessage());
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("mqtt连接失败。。" + e.getMessage());
        }
    }

然后在消息客户端回调类这里

package com.ruoyi.yyt.mqtt.config;
 
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
 
@Slf4j
@Component
public class MqttClientCallBack implements MqttCallback {

    @Value("${spring.mqtt.client.id}")
    private String clientId;
 
   /**
     * 客户端断开连接的回调
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.error(clientId + "与服务器断开连接！！" + cause.getMessage());
    }
 
    /**
     * 消息到达的回调
     */
    @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())));
        System.out.println(String.format("接收消息retained : %b",message.isRetained()));
    }
 
    /**
     * 消息发布成功的回调
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        IMqttAsyncClient client = token.getClient();
        System.out.println(client.getClientId() + "发布消息成功！");
    }
    
}

这个时候我们启动服务，调用测试接口

就可以看到接口返回发布成功，并且能看到后台服务的打印日志了

至此大功告成了！

三、参考案例

参考

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2294633.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

植物大战僵尸融合版（电脑/安卓）

植物大战僵尸融合版（电脑/安卓）

《植物大战僵尸融合版》是一款由B站UP主“蓝飘飘fly”制作的同人策略塔防游戏&#xff0c;基于经典《植物大战僵尸》玩法&#xff0c;加入了独特的植物融合系统。 出于方便&#xff0c;软件是便携版&#xff0c;解压后双击即可畅玩。 游戏主页依旧是植物大战僵尸经典界面。右下…
阅读更多...
02DevOps基础环境准备

02DevOps基础环境准备

准备两台Linux的操作系统&#xff0c;最简单的方式就是在本机上使用虚拟机搭建两个操作系统&#xff08;实际生产环境是两台服务器&#xff0c;虚拟机的方式用于学习使用&#xff09; 我搭建的两台服务器的ip分别是192.168.1.10、192.168.1.11 192.168.1.10服务器用于安装doc…
阅读更多...
苍穹外卖-day12（工作台、数据导出）

苍穹外卖-day12（工作台、数据导出）

工作台Apache POI导出运营数据Excel报表 功能实现&#xff1a;工作台、数据导出 工作台效果图&#xff1a; 数据导出效果图&#xff1a; 在数据统计页面点击数据导出&#xff1a;生成Excel报表 1. 工作台 1.1 需求分析和设计 1.1.1 产品原型 工作台是系统运营的数据看板&…
阅读更多...
详解享元模式

详解享元模式

引言 在计算机中&#xff0c;内存是非常宝贵的资源&#xff0c;而程序中可能会有大量相似或相同的对象&#xff0c;它们的存在浪费了许多空间。而享元模式通过共享这些对象&#xff0c;从而解决这种问题的。 1.概念 享元模式(Flyweight Pattern)&#xff1a;运用共享技术有效地…
阅读更多...
openEuler22.03LTS系统升级docker至26.1.4以支持启用ip6tables功能

openEuler22.03LTS系统升级docker至26.1.4以支持启用ip6tables功能

本文记录了openEuler22.03LTS将docker升级由18.09.0升级至26.1.4的过程&#xff08;当前docker最新版本为27.5.1&#xff0c;生产环境为保障稳定性&#xff0c;选择升级到上一个大版本26的最新小版本&#xff09;。 一、现有环境 1、系统版本 [rootlocalhost opt]# cat /etc…
阅读更多...
＜ OS 有关 ＞ Ubuntu 版本升级 实践 24.04 -＞ 24.10， 安装 .NET

＜ OS 有关 ＞ Ubuntu 版本升级 实践 24.04 -＞ 24.10， 安装 .NET

原因&#xff1a; 想安装 .NET 9 去编译 GitHut 项目&#xff0c;这回用不熟悉的 Ubuntu来做&#xff0c;不知道怎么拐去给 Ubuntu 升级&#xff0c;看到现在版本是 24.10 但不是 LTS 版本&#xff0c;记录下升级过程。 一、实践过程&#xff1a; 1. 查看当前版本 命令1: l…
阅读更多...
某咨询大数据解决方案介绍（32页PPT）

某咨询大数据解决方案介绍（32页PPT）

本文档介绍了一个大数据平台解决方案&#xff0c;旨在解决企业当前面临的数据问题&#xff0c;包括数据定义缺失、重复采集和存储、数据不完整以及缺乏可靠决策依据等。通过引入大数据技术&#xff0c;该方案强调从被动的IT支撑向主动的数据核心服务转型&#xff0c;以实现科学…
阅读更多...
matlab simulink 汽车四分之一模型主动被动悬架-LQR

matlab simulink 汽车四分之一模型主动被动悬架-LQR

1、内容简介 略 matlab simulink 可以交流、咨询、答疑 124- 2、内容说明 略汽车悬架系统由弹性元件、导向元件和减振器组成,是车身与车轴之间连接的所有组合体零件的总称,也是车架(或承载式车身)与车桥(或车轮)之间一切力传递装置的总称,其主要功能是使车轮与地面有很好的…
阅读更多...
从零开始：OpenCV 图像处理快速入门教程

从零开始：OpenCV 图像处理快速入门教程

文章大纲 第1章 OpenCV 概述 1.1　OpenCV的模块与功能　 1.2 OpenCV的发展 1.3 OpenCV的应用 第2章 基本数据类型 2.1 cv::Vec类 2.2 cv&#xff1a;&#xff1a;Point类 2.3 cv&#xff1a;&#xff1a;Rng类 2.4 cv&#xff1a;&#xff1a;Size类 2.5 cv&#xff1a;&…
阅读更多...
强化学习笔记6——异同策略、AC、等其他模型总结

强化学习笔记6——异同策略、AC、等其他模型总结

异步两种方法&#xff1a;1&#xff1a;经验回放 2&#xff1a;数据动作非同时产生 举例QLearning为什么是异策略&#xff1f; 生成动作时e的概率从Q表选&#xff0c;1-e概况随机。 更新策略时&#xff0c;贪心策略选择Q_max作为动作。 策略优化两种主要方法&#xff1a;基于梯…
阅读更多...
Linux提权--passwd提权

Linux提权--passwd提权

passwd​ 命令用于更改用户密码。在 Linux 系统中&#xff0c;普通用户可以通过 passwd​ 更改自己的密码&#xff0c;但如果攻击者能够以某种方式执行 passwd​ 命令更改 root 用户的密码&#xff0c;他们就能获取 root 权限。 1.常见的 passwd 提权方法 SUID 设置&#xff1…
阅读更多...
一、本地部署安装 DeepSeek 并训练本地知识库，并调用对话框进行问答

一、本地部署安装 DeepSeek 并训练本地知识库，并调用对话框进行问答

本地部署安装 DeepSeek 1、硬件环境 操作系统&#xff1a;Windows10 内存&#xff1a;16G 显卡&#xff1a;NIVIDIA GeForce RTX 2060 6G 2、安装步骤 &#xff08;1&#xff09;安装 Ollama 访问Ollama 官网&#xff0c;点击 “Download for Windows” 下载安装程序。下载…
阅读更多...
海思的一站式集成环境Hispark Studio更新了

海思的一站式集成环境Hispark Studio更新了

HiSpark Studio是海思提供的面向智能设备开发者提供一站式集成开发环境&#xff0c;支持代码编辑、编译、烧录和调试等功能。我以前在评测星闪芯片的时候用过&#xff0c;当时写了篇博客&#xff1a;【星闪开发连载】WS63E开发板Windows环境的构建_hispark studio-CSDN博客。那…
阅读更多...
unity学习29：摄像机camera相关skybox 和 Render Texture测试效果

unity学习29：摄像机camera相关skybox 和 Render Texture测试效果

目录 1 摄像机 1.1 每个Scene里都自带一个摄像机 camera 1.2 可以创建多个camera 1.3 下面先看backgroundtype: 2 backgroundtype: 天空盒 skybox 2.1 清除标志,清除&#xff1a;天空盒 自选天空盒 2.2 window /Asset Store 2.3 导入skybox 3 backgroundtype: 纯色…
阅读更多...
【Elasticsearch】Geo-distance聚合

【Elasticsearch】Geo-distance聚合

geo_distance聚合的形状是圆形。它基于一个中心点&#xff08;origin&#xff09;和一系列距离范围来计算每个文档与中心点的距离&#xff0c;并将文档分配到相应的距离范围内。这种聚合方式本质上是以中心点为圆心&#xff0c;以指定的距离范围为半径的圆形区域来划分数据。 为…
阅读更多...
音频进阶学习十二——Z变换

音频进阶学习十二——Z变换

文章目录 前言一、Z变换1.Z变换的作用2.Z变换公式3.Z的状态表示1&#xff09; r 1 r1 r12&#xff09; 0 < r < 1 0<r<1 0<r<13&#xff09; r > 1 r>1 r>1 4.关于Z的解释 二、收敛域1.收敛域的定义2.收敛域的表示方式3.ROC的分析1&#xff09;当 …
阅读更多...
easyxor

easyxor

easyxor 一、查壳 无壳&#xff0c;64位 二、IDA分析 1.main 2.查看key与r(shifee提取) 三、脚本 r [0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, …
阅读更多...
通过多层混合MTL结构提升股票市场预测的准确性，R²最高为0.98

通过多层混合MTL结构提升股票市场预测的准确性，R²最高为0.98

“Boosting the Accuracy of Stock Market Prediction via Multi-Layer Hybrid MTL Structure” 论文地址&#xff1a;https://arxiv.org/pdf/2501.09760 ​​​​​​​ 摘要 本研究引入了一种创新的多层次混合多任务学习架构&#xff0c;致力于提升股市预测的效能。此架构融…
阅读更多...
日本游戏机市场5年来首次陷入萎缩；特斯拉招人推进人形机器人量产；任天堂专利显示Switch2手柄可用作鼠标...| 游戏智眼日报

日本游戏机市场5年来首次陷入萎缩；特斯拉招人推进人形机器人量产；任天堂专利显示Switch2手柄可用作鼠标...| 游戏智眼日报

美团成立“算法顾问委员会” 美团宣布&#xff0c;近日&#xff0c;由外部专家学者组成的算法顾问委员会成立&#xff0c;为美团改进算法提供常态化咨询和指导。每个季度美团将举办算法恳谈会&#xff0c;持续邀请骑手、商家、用户、专家学者和媒体代表等共同参加。美团表示&a…
阅读更多...
114-机器学习分类算法

114-机器学习分类算法

1、内容简介 略 matlab simulink 114-机器学习分类算法可以交流、咨询、答疑 2、内容说明 略 Elong_6.24。ROCAUC confusion newdata Unbalanced_LR.car 3、仿真分析 略 4、参考论文 略
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023