Springboot实现MQTT通信的示例代码

更新时间：2025年01月24日 10:18:01 作者：就不告诉你嘿嘿

本文主要介绍了Springboot实现MQTT通信的示例代码,包含了MQTT协议的特点和工作原理等,具有一定的参考价值,感兴趣的可以了解一下

一、MQ协议

MQTT 特点

轻量级协议：
- 设计简单，占用带宽少，特别适合嵌入式设备和不稳定的网络环境。
发布/订阅模型：
- 客户端通过主题（Topic）发布消息，订阅者通过主题接收消息，彼此不直接通信。
可靠性保障：
- 提供三种服务质量（QoS）等级，确保消息可靠传输：
  - QoS 0：至多一次（不确认，可能丢失）。
  - QoS 1：至少一次（需要确认，但可能重复）。
  - QoS 2：仅一次（确保消息不丢失且不重复）。
持续连接：
- 使用 TCP/IP 连接，通过心跳包（Keep-Alive）保持连接稳定。
支持离线消息：
- 使用“保留消息”和“持久会话”功能，实现离线设备接收消息。
安全性：
- 支持 SSL/TLS 加密，结合用户名和密码进行身份验证。

MQTT 工作原理

连接：
- 客户端通过 CONNECT 消息向服务器建立连接，服务器返回 CONNACK 消息。
发布：
- 客户端通过 PUBLISH 消息向服务器发布消息，指定消息的主题。
订阅：
- 客户端通过 SUBSCRIBE 消息订阅一个或多个主题，服务器将匹配主题的消息推送给客户端。
心跳：
- 客户端和服务器定期发送心跳包（PINGREQ 和 PINGRESP），确保连接有效。
断开：
- 客户端通过 DISCONNECT 消息通知服务器主动断开连接。

MQTT 主要应用场景

物联网（IoT）：
- 设备状态监控、数据收集和远程控制。
智能家居：
- 控制家电、监控传感器数据。
车联网：
- 实时车辆数据传输、位置追踪。
移动应用：
- 消息推送、实时聊天。
工业领域：
- 设备数据采集和分析。

MQTT 配置与注意事项

主题命名：
- 使用层级结构（如 /iot/device/status），便于管理。
- 避免过于复杂的主题结构。
QoS 选择：
- 根据应用需求选择适合的 QoS 等级，平衡可靠性和性能。
安全措施：
- 启用 SSL/TLS 加密。
- 配置用户名和密码，限制匿名连接。
- 控制主题的访问权限。
性能优化：
- 控制消息大小，减少带宽占用。
- 调整心跳时间，优化连接稳定性。

二、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;

    /**
     * 客户端对象
     */
    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() + "发布消息成功！");
    }
    
}

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

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

至此大功告成了！

到此这篇关于Springboot实现MQTT通信的示例代码的文章就介绍到这了,更多相关Springboot MQTT通信内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

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

SpringBoot整合weixin-java-pay实现微信小程序支付的示例代码
微信小程序支付是常见的一种功能,本文主要介绍了SpringBoot整合weixin-java-pay实现微信小程序支付的示例代码,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习吧
2024-05-05
Java语言实现简单FTP软件 FTP软件主界面（4）
这篇文章主要为大家详细介绍了Java语言实现简单FTP软件，FTP软件主界面编写的方法，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2017-03-03
图解Java经典算法冒泡排序的原理与实现
冒泡排序是一种简单的排序算法，它也是一种稳定排序算法。其实现原理是重复扫描待排序序列，并比较每一对相邻的元素，当该对元素顺序不正确时进行交换。一直重复这个过程，直到没有任何两个相邻元素可以交换，就表明完成了排序
2022-09-09
Spring使用Configuration注解管理bean的方式详解
在Spring的世界里,Configuration注解就像是一位细心的园丁,它的主要职责是在这个繁花似锦的园子里,帮助我们声明和管理各种各样的bean,本文给大家介绍了在Spring中如何优雅地管理你的bean,需要的朋友可以参考下
2024-05-05
怎样提高mybatis-plus中saveBatch方法的效率
这篇文章主要介绍了怎样提高mybatis-plus中saveBatch方法的效率问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-07-07
如何将DeepSeek 集成到 Java 的 Spring Boot&
本文介绍了如何将DeepSeek集成到Java的SpringBoot项目中,包括准备工作、集成步骤和示例说明,感兴趣的朋友一起看看吧
2025-02-02
基于JavaMail API收发邮件的方法
这篇文章主要介绍了基于JavaMail API收发邮件的方法,实例分析了javamail的使用方法与相关注意事项,非常具有实用价值,需要的朋友可以参考下
2015-07-07
JAVA实现Excel和PDF上下标的操作代码
这篇文章主要介绍了JAVA实现Excel和PDF上下标,本文通过示例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2023-09-09
Java并发编程之synchronized底层实现原理分析
这篇文章主要介绍了Java并发编程之synchronized底层实现原理,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2024-02-02
浅谈用SpringBoot实现策略模式
本文主要介绍了SpringBoot实现策略模式，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2021-10-10