SpringBoot集成MQTT实现交互服务通信

引言

本文是springboot集成mqtt的一个实战案例。
gitee代码库地址：源码地址

一、什么是MQTT

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的“轻量级”通讯协议，该协议构建于 TCP/IP 协议上，由 IBM 于 1999 年发明。MQTT 协议的主要特征是开放、简单、轻量级和易于实现，这些特征使得它适用于受约束的应用环境，如：

网络受限：网络带宽较低且传输不可靠
终端受限：协议运行在嵌入式设备上，嵌入式终端的处理器、内存等是受限的

MQTT 非常适用于物联网领域，如传感器与服务器的通信、传感器信息采集等。

二、发布/订阅模式

发布/订阅模式（Publish/Subscribe Pattern，简称Pub/Sub）是一种消息通信模式，在这种模式下，消息的发送者（发布者）不会将消息直接发送给特定的接收者（订阅者）。而是将代表消息内容的通知（事件）发布到一个特定的主题或频道上，而订阅了这个主题的接收者会收到所有在这个主题上发布的通知。这种模式解耦了消息的发送者和接收者，使得系统更加灵活和可扩展。

主要组成部分

  1. 发布者（Publisher）：负责生成消息并将其发布到特定的主题或频道。

  2. 订阅者（Subscriber）：注册对特定主题的兴趣，并接收该主题上的所有消息。

  3. 消息代理（Message Broker）：作为中间件，它接收来自发布者的消息，并将这些消息传递给所有相关的订阅者。

优点

  • 解耦：发布者和订阅者之间不需要直接交互，这降低了系统的耦合度。

  • 灵活性：可以动态添加或删除订阅者，不影响其他组件。

  • 可扩展性：系统容易扩展，可以轻松增加新的发布者或订阅者。

缺点

  • 复杂性：引入了额外的组件（如消息代理），增加了系统的复杂性和管理成本。

  • 性能开销：消息的传递需要通过中间件，可能会有延迟和性能损失。

应用场景

  • 事件驱动架构：在微服务架构中，不同的服务通过发布/订阅模式进行异步通信。

  • 数据流处理：如实时数据分析，多个组件可以订阅数据流并进行处理。

  • 分布式系统：用于跨系统或跨服务的消息传递。

发布/订阅模式并不是 MQTT 协议特有的模式，很多消息中间件都有使用发布/订阅模式，有同学可能认为这就是观察者模式，还真不是，这两个模式很容易混淆。观察者模式只有观察者 + 被观察者两个角色，而发布/订阅模式还有一个经纪人 Broker；往更深层次的讲观察者和被观察者，是松耦合的关系，而发布者和订阅者，则完全不存在耦合。

三、Windows下安装MQTT消息服务器

非常遗憾，EMQ X Broker 在 5.4.0 版本的发行版中已不支持 windows 版本的安装包了，笔者从网上找了一个最后支持版本的压缩包，已上传资源。

  • 解压后，在bin文件下，使用cmd执行运行命令 .\emqx console
  • 访问MQTT管理页面 http://localhost:18083/#/ 用户名密码 admin/public

如果报错缺少Erlang环境，需要自行安装下该环境

在这里插入图片描述
在这里插入图片描述

浏览器访问：http://localhost:18083/#，输入账号密码进入，会要求你修改密码，可以暂时跳过

在这里插入图片描述

四、Windows安装MQTT消息代理客户端MQTTX

下载地址：MQTTX下载地址

点击免费下载
在这里插入图片描述

选择64位版本

在这里插入图片描述
下好后点击安装，启动运行界面如下：
在这里插入图片描述
语言是英文，可以在设置按钮里调成中文。这个客户端代理主要是进行消息发送的测试服务。

五、新建MQTT集成项目

随便新建了一个springboot应用，用的是JDK17，在pom文件中引入如下依赖：

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

5.1 yml配置

server:port: 8081#允许循环依赖
spring:main:allow-circular-references: truecustomer:mqtt:broker: tcp://localhost:1883clientList:#发布客户端ID- clientId: nays_service#监听主题 同时订阅多个主题 使用 - 分割开subscribeTopic: mqtt/publish#用户名userName: admin#密码password: public#接收客户端ID- clientId: receive_service#监听主题 同时订阅多个主题 使用 - 分割开subscribeTopic: mqtt/receive#用户名userName: admin#密码password: public

5.2 Mqtt配置类

package com.hulei.mqttproject.config;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;import java.util.List;/*** Mqtt配置类*/
@Data
@Configuration
@ConfigurationProperties(prefix = "customer.mqtt")
public class MqttConfig {/*** mqtt broker地址*/String broker;/*** 需要创建的MQTT客户端*/List<MqttClient> clientList;
}

5.3 MQTT客户端

package com.hulei.mqttproject.config;import lombok.Data;/*** MQTT客户端*/
@Data
public class MqttClient {/*** 客户端ID*/private String clientId;/*** 监听主题*/private String subscribeTopic;/*** 用户名*/private String userName;/*** 密码*/private String password;
}

5.4 MQTT客户端管理类

package com.hulei.mqttproject.config;import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** MQTT客户端管理类,如果客户端非常多后续可入redis缓存*/
@Slf4j
@Component
public class MqttClientManager {@Value("${customer.mqtt.broker}")private String mqttBroker;@Resourceprivate MqttCallBackContext mqttCallBackContext;/*** 存储MQTT客户端*/public static Map<String, MqttClient> MQTT_CLIENT_MAP = new ConcurrentHashMap<>();public MqttClient getMqttClientById(String clientId) {return MQTT_CLIENT_MAP.get(clientId);}/*** 创建mqtt客户端** @param clientId       客户端ID* @param subscribeTopic 订阅主题，可为空* @param userName       用户名，可为空* @param password       密码，可为空*/public void createMqttClient(String clientId, String subscribeTopic, String userName, String password) {MemoryPersistence persistence = new MemoryPersistence();try {MqttClient client = new MqttClient(mqttBroker, clientId, persistence);MqttConnectOptions connOpts = new MqttConnectOptions();if (null != userName && !userName.isEmpty()) {connOpts.setUserName(userName);}if (null != password && !password.isEmpty()) {connOpts.setPassword(password.toCharArray());}connOpts.setCleanSession(true);if (null != subscribeTopic && !subscribeTopic.isEmpty()) {AbsMqttCallBack callBack = mqttCallBackContext.getCallBack(clientId);if (null == callBack) {callBack = mqttCallBackContext.getCallBack("default");}callBack.setClientId(clientId);callBack.setConnectOptions(connOpts);client.setCallback(callBack);}//连接mqtt服务端brokerclient.connect(connOpts);// 订阅主题if (null != subscribeTopic && !subscribeTopic.isEmpty()) {if (subscribeTopic.contains("-"))client.subscribe(subscribeTopic.split("-"));else {client.subscribe(subscribeTopic);}}MQTT_CLIENT_MAP.putIfAbsent(clientId, client);} catch (MqttException e) {log.error("Create mqttClient failed!", e);}}
}

5.5 MQTT客户端创建

package com.hulei.mqttproject.config;import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import java.util.List;/*** MQTT客户端创建*/
@Component
@Slf4j
public class MqttClientCreate {@Resourceprivate MqttClientManager mqttClientManager;@Resourceprivate MqttConfig mqttConfig;/*** 创建MQTT客户端*/@PostConstructpublic void createMqttClient() {List<MqttClient> mqttClientList = mqttConfig.getClientList();for (MqttClient mqttClient : mqttClientList) {log.info("{}", mqttClient);//创建客户端，客户端ID：demo，回调类跟客户端ID一致mqttClientManager.createMqttClient(mqttClient.getClientId(), mqttClient.getSubscribeTopic(), mqttClient.getUserName(), mqttClient.getPassword());}}
}

5.6 MQTT回调抽象类

package com.hulei.mqttproject.config;import jakarta.annotation.Resource;
import lombok.Getter;
import lombok.Setter;
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.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;/*** MQTT回调抽象类*/
@Setter
@Getter
@Slf4j
public abstract class AbsMqttCallBack implements MqttCallback {private String clientId;private MqttConnectOptions connectOptions;@ResourceMqttClientManager mqttClientManager;/*** 失去连接操作,进行重连** @param throwable 异常*/@Overridepublic void connectionLost(Throwable throwable) {try {if (null != clientId) {if (null != connectOptions) {mqttClientManager.getMqttClientById(clientId).connect(connectOptions);} else {mqttClientManager.getMqttClientById(clientId).connect();}}} catch (Exception e) {log.error("{} reconnect failed!", e.getMessage(), e);}}/*** 接收订阅消息* @param topic    主题* @param mqttMessage 接收消息* @throws Exception 异常*/@Overridepublic void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {String content = new String(mqttMessage.getPayload());handleReceiveMessage(topic, content);}/*** 消息发送成功** @param iMqttDeliveryToken toke*/@Overridepublic void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {log.info("消息发送成功");}/*** 处理接收的消息* @param topic   主题* @param message 消息内容*/protected abstract void handleReceiveMessage(String topic, String message);
}

5.7 MQTT订阅回调环境类

package com.hulei.mqttproject.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** MQTT订阅回调环境类*/
@Component
@Slf4j
public class MqttCallBackContext {private final Map<String, AbsMqttCallBack> callBackMap = new ConcurrentHashMap<>();/*** 默认构造函数** @param callBackMap 回调集合*/public MqttCallBackContext(Map<String, AbsMqttCallBack> callBackMap) {this.callBackMap.putAll(callBackMap);}/*** 获取MQTT回调类** @param clientId 客户端ID* @return MQTT回调类*/public AbsMqttCallBack getCallBack(String clientId) {return this.callBackMap.get(clientId);}
}

5.8 默认回调类

package com.hulei.mqttproject.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;/*** 默认回调*/
@Slf4j
@Component("default")
public class DefaultMqttCallBack extends AbsMqttCallBack {/*** @param topic   主题* @param message 消息内容*/@Overrideprotected void handleReceiveMessage(String topic, String message) {log.info("接收到主题---{}", topic);log.info("接收到消息---{}", message);// 自定义消息处理业务}
}

六、测试服务类

package com.hulei.mqttproject.controller;import com.hulei.mqttproject.config.MqttClientManager;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
@Slf4j
public class SendController {@Resourceprivate MqttClientManager mqttClientManager;@RequestMapping("/sendMessage")public String sendMessage(String topic){try {MqttMessage mqttMessage = new MqttMessage("你好".getBytes());mqttClientManager.getMqttClientById("nays_service").publish(topic,mqttMessage);return "发送成功";} catch (Exception e) {log.error("发送失败",e);return "发送失败";}}
}

七、启动springboot

启动日志可以看到，mqtt消息服务器连接成功

在这里插入图片描述
EMQX工具显示发布客户端和接收客户端均已成功注册

在这里插入图片描述

使用Apifox测试下SendController中的接口，mqtt/receive是yaml中接收客户端订阅的主题，当然也可以往mqtt/publish主题发，mqtt中消息的发布者也可以订阅主题，监听某些消息。

在这里插入图片描述

在这里插入图片描述

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

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

教大模型学数学，总共分几步？

教大模型学数学，总共分几步？

大模型那么聪明&#xff0c;为什么数学题总是做不对、做不会&#xff1f; 从答高考数学卷难及格到普通数字比大小出错&#xff0c;大模型总算让大家觉得并非“无所不能”。这一方面让普通人开心&#xff0c;毕竟讲到AI取代人类看起来为时尚早&#xff0c;而另一方面&#xff0…
阅读更多...
Autosar RTE配置-Assembly和Delegation的使用-基于ETAS软件

Autosar RTE配置-Assembly和Delegation的使用-基于ETAS软件

文章目录 前言Assembly和Delegation的含义Delegation的使用Assembly的使用总结 前言 RTE中的Compostion内部的SWC之间的连接使用Assembly Connector进行连接。这样的连接一般都是一个SWC的Pport对应另一个SWC的Rport。而Autosar软件中往往不只一个Composition(一般可以以核的数…
阅读更多...
Android Toast

Android Toast

Toast Toast是Android常用的简单控件&#xff0c;主要用来进行简短的信息提示&#xff0c;如图1所示。 图1 Toast效果图 Toast的基本用法很简单&#xff0c;不需要设置layout&#xff0c;只需要在程序中调用即可。Toast调用makeText()方法设置需要显示的界面、显示的内容、显…
阅读更多...
【计算机网络】学习指南及导论

【计算机网络】学习指南及导论

个人主页&#xff1a;【&#x1f60a;个人主页】 系列专栏&#xff1a;【❤️计算机网络】 文章目录 前言我们为什么要学计算机网络&#xff1f;计算机网络概述计算机网络的分类按交换技术分类按使用者分类按传输介质分类按覆盖网络分类按覆盖网络分类 局域网的连接方式有线连接…
阅读更多...
Lua基础知识入门

Lua基础知识入门

1 基础知识 标识符&#xff1a;标识符的定义和 C语言相同&#xff1a;字母和下划线_ 开头&#xff0c; 下划线_ 大写字母一般是lua保留字&#xff0c; 如_VERSION 全局变量&#xff1a;默认情况下&#xff0c;变量总是认为是全局的&#xff0c;不需要申明&#xff0c;给一个变…
阅读更多...
河南萌新联赛2024第（一）场：河南农业大学

河南萌新联赛2024第（一）场：河南农业大学

A.造数 题目&#xff1a; 链接&#xff1a;https://ac.nowcoder.com/acm/contest/86639/A 思路&#xff1a; 签到题&#xff0c;特判如果n0&#xff0c;输出0&#xff0c;如果n1或2&#xff0c;输出1&#xff1b;while循环&#xff0c;首先如果n%2!0&#xff0c;那么s&…
阅读更多...
基于Web的特产美食销售系统的设计与实现

基于Web的特产美食销售系统的设计与实现

&#x1f497;博主介绍&#x1f497;&#xff1a;✌在职Java研发工程师、专注于程序设计、源码分享、技术交流、专注于Java技术领域和毕业设计✌ 温馨提示&#xff1a;文末有 CSDN 平台官方提供的老师 Wechat / QQ 名片 :) Java精品实战案例《700套》 2025最新毕业设计选题推荐…
阅读更多...
Spring Web MVC（常用的注解@RequestMapping,@RequestParam,@RequestBody等）

Spring Web MVC（常用的注解@RequestMapping,@RequestParam,@RequestBody等）

一、Spring MVC spring的启动类 启动类是看这个 SpringBootApplication 注解&#xff0c;而不是 类的名字 这个注解在哪&#xff0c;哪个类就是启动类 1.MVC思想 举例 二、Spring MVC mvc 是一种思想&#xff0c;而spring mvc是对mvc思想的一种实现。全称是 spring web mvc…
阅读更多...
笔记 2 ： 课本第 3 章开始，记录 arm 的汇编指令的格式

笔记 2 ： 课本第 3 章开始，记录 arm 的汇编指令的格式

&#xff08;13&#xff09; 介绍 arm 中的第一个汇编指令的用法 mov &#xff1a; &#xff08;14&#xff09;立即数的概念&#xff1a; &#xff08;15&#xff09; 汇编中的移位写法&#xff1a; 举例 &#xff1a; &#xff08;16&#xff09; 学习一个新的指令 cmp &a…
阅读更多...
【大型实战】企业网络实验(华为核心交换、ESXI7.0vmware虚拟机、DHCP中继、服务端网络及用户端网络配置)

【大型实战】企业网络实验(华为核心交换、ESXI7.0vmware虚拟机、DHCP中继、服务端网络及用户端网络配置)

需求 实验 vmware网络配置&#xff08;企业内部一般为ESXI&#xff09; 这样服务器虚拟机使用192.168.200.X网段才能与用户侧互通 vmware虚拟机配置&#xff08;DHCP服务器网络配置&#xff09; 打开网络管理页面 nmtui重置一下网络连接&#xff08;重启网卡&#xff09; …
阅读更多...
ubuntu24.04 linux bcm94352hmb 无线网卡（带蓝牙功能）无法连接蓝牙设备的解决办法

ubuntu24.04 linux bcm94352hmb 无线网卡（带蓝牙功能）无法连接蓝牙设备的解决办法

ubuntu24.04 linux bcm94352hmb 无线网卡&#xff08;带蓝牙功能&#xff09;无法连接蓝牙设备的解决办法 问题描述 小本是自己换装的bcm94352hmb&#xff0c;现在跑不动黑苹果了&#xff0c;换装ubuntu&#xff0c;该网卡在黑苹果上和windows上都能正常工作&#xff0c;到了…
阅读更多...
Go语言中的并发

Go语言中的并发

简单介绍go中的并发编程. 涉及内容主要为goroutine, goroutine间的通信(主要是channel), 并发控制(等待、退出). 想查看更多与Go相关的内容, 可以查看我的Go编程栏目 Goroutine 语法 在一个函数调用前加上go即可, go func(). 语法很简单, 可以说是并发写起来最简单的程序语言…
阅读更多...
Mongodb多键索引中索引边界的混合

Mongodb多键索引中索引边界的混合

学习mongodb&#xff0c;体会mongodb的每一个使用细节&#xff0c;欢迎阅读威赞的文章。这是威赞发布的第93篇mongodb技术文章&#xff0c;欢迎浏览本专栏威赞发布的其他文章。如果您认为我的文章对您有帮助或者解决您的问题&#xff0c;欢迎在文章下面点个赞&#xff0c;或者关…
阅读更多...
linux进程——概念理解与PCB

linux进程——概念理解与PCB

前言&#xff1a;本篇讲解Linux进程概念相关内容。 操作系统被称为计算机世界的哲学&#xff0c; 可以见得操作系统的知识并不好理解。 对于这篇进程概念的讲解&#xff0c; 博主认为&#xff0c; 如果没有一些前置知识的话&#xff0c;里面的有些概念并不好理解。 但是如果学习…
阅读更多...
TDesign组件库日常应用的一些注意事项

TDesign组件库日常应用的一些注意事项

【前言】Element&#xff08;饿了么开源组件库&#xff09;在国内使用的普及率和覆盖率高于TDesign-vue&#xff08;腾讯开源组件库&#xff09;&#xff0c;这也导致日常开发遇到组件使用上的疑惑时&#xff0c;网上几乎搜索不到其文章解决方案&#xff0c;只能深挖官方文档或…
阅读更多...
防溺水预警系统引领水域安全新篇章

防溺水预警系统引领水域安全新篇章

一、系统概述 随着人们对水域活动的需求增加&#xff0c;溺水事故频发&#xff0c;给人们的生命安全带来了严重威胁。然而&#xff0c;如今&#xff0c;一项创新科技正在以强大的功能和无限的潜力引领着水域安全的新篇章。智能防溺水预警系统&#xff0c;作为一种集成了智能感知…
阅读更多...
文心一言：如何备考软考架构师

文心一言：如何备考软考架构师

提问&#xff1a;软考高级:架构师指导&#xff0c;有哪些考点、重点考点有哪些&#xff1f;需要如何备考&#xff1f;案例分析题如何做&#xff1f;论文如何写&#xff1f; 文心一言回答1700字如下&#xff1a; 针对软考高级架构师的考试&#xff0c;以下是对考点、重点考点、备…
阅读更多...
MongoDB自学笔记（三）

MongoDB自学笔记（三）

一、前文回顾 上一篇文章中我们学习了更新操作&#xff0c;以及讲解了部分的更新操作符&#xff0c;今天我们继续学习剩余的更新操作符。 二、更新操作符 1、$rename 语法&#xff1a;{ $rename: { < field1 >: < newName1 >, < field2 >: < newName2…
阅读更多...
JDK、JRE、JVM的区别java的基本数据类型

JDK、JRE、JVM的区别java的基本数据类型

说一说JDK、JRE、JVM的区别在哪&#xff1f; JDK&#xff1a; Java Delopment kit是java工具包&#xff0c;包含了编译器javac&#xff0c;调试器&#xff08;jdb&#xff09;以及其他用于开发和调试java程序的工具。JDK是开发人员在开发java应用程序时候所需要的的基本工具。…
阅读更多...
Linux - 基础开发工具（yum、vim、gcc、g++、make/Makefile、git、gdb）

Linux - 基础开发工具（yum、vim、gcc、g++、make/Makefile、git、gdb）

目录 Linux软件包管理器 - yum Linux下安装软件的方式 认识yum 查找软件包 安装软件 如何实现本地机器和云服务器之间的文件互传 卸载软件 Linux编辑器 - vim vim的基本概念 vim下各模式的切换 vim命令模式各命令汇总 vim底行模式各命令汇总 vim的简单配置 Linux编译器 - gc…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023