导航：首页 > 开发技术 >

Springboot中如何使用MQTT

发表于：2026-11-04 作者：千家信息网编辑

千家信息网最后更新 2026年11月04日，这篇文章主要为大家展示了"Springboot中如何使用MQTT"，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下"Springboot中如何使用MQTT"这篇

千家信息网最后更新 2026年11月04日Springboot中如何使用MQTT

这篇文章主要为大家展示了"Springboot中如何使用MQTT"，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下"Springboot中如何使用MQTT"这篇文章吧。

为什么选择MQTT

MQTT的定义相信很多人都能讲的头头是道,本文章也不讨论什么高大上的东西,旨在用最简单直观的方式让每一位刚接触的同行们可以最快的应用起来

先从使用MQTT需要什么开始分析:

消息服务器
不同应用/设备之间的频繁交互
可能涉及一对多的消息传递

根据上面列举的这三点,我们大概可以了解到, MQTT最适合的场景是消息做为系统的重要组成部分,且参与着系统关键业务逻辑的情形

MQTT, 启动!

既然决定使用它,我们首先要研究的是如何让MQTT正常工作,毕竟它不是简单的在maven里加入个依赖就完事的

我们总共需要干如下两件事:

下载EMQX消息服务器, 作为broker
在maven中引入依赖

      org.springframework.integration      spring-integration-mqtt      5.3.2.RELEASE

完成上面两步后, 启动EMQX服务器, 正式进入我们的MQTT旅途

使用方式

在Spring Boot中使用MQTT的代码, 笔者总结了如下两种方式:

使用spring-integration的消息通道概念
使用传统的Client客户端概念

第一种会产生一定程度的心智负担,但在笔者成功搭配(抄袭+造轮子)自动注册后, 比后者要方便许多

在介绍具体代码之前, 我们先简单整理下使用中最常见的概念:

主题: MQTT消息的主要传播途径, 我们向主题发布消息, 订阅主题, 从主题中读取消息并进行业务逻辑处理, 主题是消息的通道
生产者: MQTT消息的发送者, 他们向主题发送消息
消费者: MQTT消息的接收者, 他们订阅自己需要的主题, 并从中获取消息
broker: 消息转发器, 消息是通过它来承载的, EMQX就是我们的broker, 在使用中我们不用关心它的具体实现

其实, MQTT的使用流程就是: 生产者给主题发消息->broker进行消息的传递->订阅该主题的消费者拿到消息并进行相应的业务逻辑

Client模式

本模式和传统的数据库链接,Redis链接基本一致,有开发经验的小伙伴们可以很轻松的驾驭,我们需要考虑的就是如果创建对应的工厂,是单例模式,还是原型,亦或是造个池子呢?

我们使用单例模式来进行本次的介绍

创建工厂类

首先, 我们创造一个工厂(就不承认设计模式中毒)

public class MqttFactory {      private static MqttProperties configuration;          private static MqttClient client;       /**    *   获取客户端实例    *   单例模式, 存在则返回, 不存在则初始化    */    public static MqttClient getInstance() {            if (client == null) {                  init();            }            return client;      }          /**    *   初始化客户端    */    public static void init() {            try {                  client = new MqttClient(configuration.getAddress(), "client-" + System.currentTimeMillis());                  // MQTT配置对象            MqttConnectOptions options = new MqttConnectOptions();                  // 设置自动重连, 其它具体参数可以查看MqttConnectOptions            options.setAutomaticReconnect(true);                  if (!client.isConnected()) {                    client.connect(options);                  }            } catch (MqttException e) {                  LOGGER.error(String.format("MQTT: 连接消息服务器[%s]失败", configuration.getAddress()));            }      }    }

关于MQTT的具体配置可以查看MqttConnectOptions, 在这里就不做说明了

多嘴一句, 文档永远比某些博客给力!!!

创建工具类

接下来, 我们创建MqttUtil, 用于消息的发送以及主题的订阅

public class MqttUtil {      /**    *   发送消息    *   @param topic 主题    *   @param data 消息内容    */    public static void send(String topic, Object data) {            // 获取客户端实例        MqttClient client = MqttFactory.getInstance();            ObjectMapper mapper = new ObjectMapper();            try {            // 转换消息为json字符串            String json = mapper.writeValueAsString(data);                  client.publish(topic, new MqttMessage(json.getBytes(StandardCharsets.UTF_8)));            } catch (JsonProcessingException e) {                  LOGGER.error(String.format("MQTT: 主题[%s]发送消息转换json失败", topic));            } catch (MqttException e) {                  LOGGER.error(String.format("MQTT: 主题[%s]发送消息失败", topic));            }      }        /**     * 订阅主题     * @param topic 主题     * @param listener 消息监听处理器     */    public static void subscribe(String topic, IMqttMessageListener listener) {          MqttClient client = MqttFactory.getInstance();          try {                client.subscribe(topic, listener);          } catch (MqttException e) {                LOGGER.error(String.format("MQTT: 订阅主题[%s]失败", topic));          }    }    }

相信小伙伴们注意到了IMqttMessageListener这个东西, 我们只需要创建一个监听类, 实现IMqttMessageListener接口, 就可以处理消息啦, 代码如下:

public class MessageListener implements IMqttMessageListener {      /**     * 处理消息    * @param topic 主题     * @param mqttMessage 消息     */    @Override      public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {           LOGGER.info(String.format("MQTT: 订阅主题[%s]发来消息[%s]", topic, new String(mqttMessage.getPayload())));      }        public static void main(String[] args) {          //订阅主题test01, 使用MessageListener来处理它的消息        MqttUtil.subscribe("test01", new MessageListener());    }}

无论是发送还是订阅,是不是都很好理解?

舒服的事情结束后, 带来的是无尽的折磨和空虚, 来吧, 让我们挑战下心智负担大的第二种模式!