Java MQTT实战应用

2025-06-27 17:50
文章标签 java 实战 应用 mqtt

本文主要是介绍Java MQTT实战应用，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

《JavaMQTT实战应用》本文详解MQTT协议,涵盖其发布/订阅机制、低功耗高效特性、三种服务质量等级（QoS0/1/2）,以及客户端、代理、主题的核心概念,最后提供Linux部署教程、Sprin...

一、MQTT协议

MQTT（Message Queuing Telemetry Transport）是一种轻量级的发布/订阅式消息传递协议，专为物联网(IoT)和嵌入式设备设计，它简化了设备之间的通信，并优化带宽使用。 

在MQTT中，消息的发送者称为“发布者”（Publisher）消息的接收者称为“订阅者”（Subscriber），而消息的中转站是“代理”（Broker）。发布者将消息发布到特定的“主题”（Topic），代理负责将消息转发给所有订阅了该主题的订阅者。这种模式解耦了消息的发送者和接收者，使得系统更加灵活和可扩展。

二、MQTT优点

  • 低功耗、高效、可靠。
  • 轻量级：协议设计简洁，消息头部开销小，适用于低带宽和低功耗设备。
  • 支持发布/订阅模式：设备可以发布消息到主题，其他设备可以订阅对应的主题接收消息。这一模式解耦了消息生产者和消费者，简化了系统架构，提高了灵活性和可扩展性。
  • 可拓展性和兼容性：MQTT允许使用不同的传输协议，包括TCP、WebSocket等。它的简单性使得它易于与其他协议和服务集成。
  • 持久化会话：MQTT支持消息持久化，允许设备在断线后重新连接时恢复之前的会话状态，包括未完成的订阅和未收到的消息队列，这对于网络不稳定或经常断开的物联网环境尤为重要。

三、三种服务质量等级

  • QoS = 0（最多一次）：消息最多被传递一次，可能丢失，但不会重复。此级别提供的可靠性最低，一旦消息被客户端发送出去，它不会等待任何确认，即“Fire and Forget”模式。这意味着发布者不会确认消息是否到达Broker，也不会尝试重传失败的消息）
  • QoS = 1（至少一次）：消息至少被传递一次，可能会重复，但不会丢失。此级别保证消息至少被送达一次，但有可能被重复发送。在QoS 1下，Broker(消息队列服务器)会发送PUBACK确认消息给客户端，如果客户端没有收到确认，则会重发消息，直到收到确认为止。因此，虽然可以确保消息不会丢失，但也可能导致相同消息被多次接收
  • QoS = 2（恰好一次）：消息保证被传递一次且仅一次，不会丢失也不会重复。这是MQTT提供的最高级别服务质量，确保每条消息只会被接收一次，提供最严格的可靠性保证。该机制通过一个复杂的四次握手过程实现，包括消息标识符的确认和释放，确保消息既不丢失也不重复

四、客户端、代理、主题

MQTT协议中编程，三个核心概念分别是客户端（Client）、代理（Broker）和主题（Topic），它们共同构成了MQTT通信的基础框架，实现了消息的发布与订阅机制。

1. 客户端（Client）：

作用：客户端可以是消息的发布者（Publisher）或订阅者（Subscriber），也可以同时具备这两种角色。发布者负责向MQTT系统中的某个主题发布消息；订阅者则订阅感兴趣的主题，以接收来自该主题的消息。客户端可以是传感器、手机应用、服务器程序等各种设备或应用。

相互关系：客户端不直接相互通信，而是通过Broker中转消息。发布者客户端向Broker发送消息，而订阅者客户端从Broker接收消息。

 2. 代理（Broker）：

作用：Broker是MQTT通信的中心节点，它接收来自发布者客户端的消息，并根据消息中的主题分发给相应的订阅者客户端。Broker负责维护客户端的连接状态、存储消息（如果需要持久化）、管理主题的订阅关系等。

相互关系：Broker是客户端之间的中介，它管理着所有的消息流动。每个客户端都与Broker建立连接，无论发布还是订阅操作，都必须通过Broker来完成。

3. 主题（Topic）：

作用：主题是MQTT中消息的分类标签，类似于一个消息通道或者频道。每个消息都会关联一个主题，发布者通过指定主题来决定消息的去向，而订阅者通过订阅特定主题来接收相关消息。

相互关系：主题是连接发布者与订阅者的桥梁。发布者向特定主题发布消息，而订阅者则通过订阅这些主题来接收消息。Broker根据主题匹配规则，确保消息被正确地路由到已订阅该主题的所有客户端。主题可以是静态的字符串，也可以包含通配符（如"+“和”#”）来实现灵活的匹配规则。

五、实战应用

1. 安装部署(linux)

 -- 拉取镜像

docker pull emqx/emqx:5.0.26

-- 安装容器

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

2. 访问控制台

访问：ip:18083

默认的用户名密码：admin/public

3. 客户端认证

Java MQTT实战应用

4. 创建用户

Java MQTT实战应用

Java MQTT实战应用

5. SpringBoot中整合

5.1 导入jar包
<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-mqtt</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-configuration-processor</artifactId>
  <optional>true</optional>
</dependency>
5.2 yml配置
mqtt:
  #MQTT-服务器连接地址，如果有多个，用逗号隔开
  host: tcp://192.168.17.101:1883
  #MQTT-连接服务器默认客户端ID，可以随便写
  clientId: mqtt_test
  #MQTT-用户名
  username: zhangsan
  #MQTT-密码
  password: 123456
  #MQTT-指定消息的推送和订阅主题
  topic: test
  #连接超时
  timeout: 100
  #设置会话心跳时间
  keepalive: 10
5.3 MqttConfig.Java
@Slf4j
@Configuration
@ConfigurationProperties("mqtt")
@Data
public class MqttConfig {
    String host;
    String clientId;
    String topic;
    String username;
    String password;
    Integer timeout;
    Integer keepalive;
    // MQTT客户端的配置类，可以设置mqtt服务器的账号和密码
    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        // 设置是否自动重连
        options.setAutomaticReconnect(true);
        // false 保持会话不被清理自动重连后才能收到订阅的主题消息（包括离线时发布的消息）
        options.setCleanSession(true);
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        return options;
    }
    // MqttClient 类，MQTT的客户端类，可以去连接MQTT服务器
    @Bean
    public MqttClient mqttClient(MqttConnectOptions mqttConnectOptions) {
        try {
            MqttClient client = new MqttClient(host, clientId);
            // 回调对象，监听消息的获取，采用的接口回调，可以获取对应订阅到的消息
            client.setCallback(new MessageCallback(client, this.topic, mqttConnectOptions));
            // 连接
            client.connect(mqttConnectOptions());
            return client;
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("mqtt 连接异常");
        }
    }
}
5.4 MessageCallback.java
/**
 * consumer 消费者，对收到的消息进行处理
 */
//@Component
@Slf4j
public class MessageCallback implements MqttCallbackExtended {
    private MqttClient client;
    private String topic;
    private MqttConnectOptions mqttConnectOptions;
    public MessageCallback() {
    }
    public MessageCallback(MqttClient mqttClient, String topic, MqttConnectOptions mqttConnectOptions) {
        this.client = mqttClient;
        this.topic = topic;
        this.mqttConnectOptions = mqttConnectOptions;
    }
    // 在客户端连接断开时触python发
    @Override
    public void connectionLost(Throwable throwable) {
        if (client != null && !client.isConnected()) {
            log.info("{}, 连接断开，正在reconnect....", client.getClientId());
            try {
                client.reconnect();
                // client.connect(this.mqttConnectOptions);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        } else {
            log.info("未知异常，连接断开");
        }
    }
    // 在客户端与服务器连接成功时触发
    @Override
    public void connectComplete(boolean b, String url) {
        log.info("{} 上线了{} {}", client.getClientId(), b, url);
        try {
            client.subscribe(this.topic, 0);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    // 在客户端收到订阅的消息时触发
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("接收消息主题 : " + topic);
        log.info("接收消息内容 : " + new String(message.getPayload()));
        String msg = new String(message.getPayload());
        try {
            jsONObject jsonObject = JSON.parseobject(msg);
            String clientId = String.valueOf(jsonObject.get("clientid"));
            if (topic.endsWith("disconnected")) {
                log.info("设备{}已掉线", clientId);
            } else if (topic.endsWith("connected")) {
                log.info("设备{}已上线", clientId);
            } else {
                log.info("其他主题的消息");
            }
        } catch (JSONException e) {
            log.error("JSON Format Parsing Exception : {}", msg);
        }
    }
    // 在客户端发送消息至服务器成功时触发
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("deliveryComplpythonete---------" + token.isComplete());
    }
}
5.5 MqttUtil.java
@Component
@Slf4j
public class MqttUtil {
    @Autowired(required = false)
    private MqttClient client;
    /**
     * 订阅主题
     *
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    /**
     * 订阅主题
     *
     * @param topic
     */
    public void subscribe(String topic) {
        try {
            client.subscribe(topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    /**
     * 发布消息
     *
     * @param qos         连接方式 0,1,2 默认0
     * @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 mqttTopic = client.getTopic(topic);
        if (null == mqttTop编程ic) {
            log.error("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            // 发送消息
            token = mqttTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    /**
     * 发布消息
     *
     * @param topic       主题
     * @param pushMessage 消息内容
     */
    public void publish(String topic, String pushMessage) {
        publish(0, true, topic, pushMessage);
    }
}
5.6 MqttController.java
@RestController
@Slf4j
public class MqttController {
    @Autowired
    MqttClient client;
    @Autowired
    MqttUtil mqttUtil;
    @GetMapping("/send")
    public String send() {
        try {
            for (int i = 0; i < 3; i++) {
                mqttUtil.publish("test", "消息hello" + i);
                log.info("发送成China编程功：{}", i);
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "SUCCESS";
    }
}

六、MQTTX官网地址

MQTT客户端工具MQTTX下载地址 : MQTTX：全功能 MQTT 客户端工具

到此这篇关于Java MQTT实战应用的文章就介绍到这了,更多相关Java MQTT内容请搜索编程China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持China编程(www.chinasem.cn)！

这篇关于Java MQTT实战应用的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1155237。

相关文章

Java中流式并行操作parallelStream的原理和使用方法

Java中流式并行操作parallelStream的原理和使用方法

《Java中流式并行操作parallelStream的原理和使用方法》本文详细介绍了Java中的并行流（parallelStream）的原理、正确使用方法以及在实际业务中的应用案例,并指出在使用并行流... 目录Java中流式并行操作parallelStream0. 问题的产生1. 什么是parallelS
阅读更多...
Java中Redisson 的原理深度解析

Java中Redisson 的原理深度解析

《Java中Redisson的原理深度解析》Redisson是一个高性能的Redis客户端,它通过将Redis数据结构映射为Java对象和分布式对象,实现了在Java应用中方便地使用Redis,本文... 目录前言一、核心设计理念二、核心架构与通信层1. 基于 Netty 的异步非阻塞通信2. 编解码器三、
阅读更多...
SpringBoot基于注解实现数据库字段回填的完整方案

SpringBoot基于注解实现数据库字段回填的完整方案

《SpringBoot基于注解实现数据库字段回填的完整方案》这篇文章主要为大家详细介绍了SpringBoot如何基于注解实现数据库字段回填的相关方法,文中的示例代码讲解详细,感兴趣的小伙伴可以了解... 目录数据库表pom.XMLRelationFieldRelationFieldMapping基础的一些代
阅读更多...
一篇文章彻底搞懂macOS如何决定java环境

一篇文章彻底搞懂macOS如何决定java环境

《一篇文章彻底搞懂macOS如何决定java环境》MacOS作为一个功能强大的操作系统,为开发者提供了丰富的开发工具和框架,下面：本文主要介绍macOS如何决定java环境的相关资料,文中通过代码... 目录方法一：使用 which命令方法二：使用 Java_home工具（Apple 官方推荐）那问题来了，
阅读更多...
Java HashMap的底层实现原理深度解析

Java HashMap的底层实现原理深度解析

《JavaHashMap的底层实现原理深度解析》HashMap基于数组+链表+红黑树结构,通过哈希算法和扩容机制优化性能,负载因子与树化阈值平衡效率,是Java开发必备的高效数据结构,本文给大家介绍... 目录一、概述：HashMap的宏观结构二、核心数据结构解析1. 数组（桶数组）2. 链表节点（Node
阅读更多...
Java AOP面向切面编程的概念和实现方式

Java AOP面向切面编程的概念和实现方式

《JavaAOP面向切面编程的概念和实现方式》AOP是面向切面编程,通过动态代理将横切关注点（如日志、事务）与核心业务逻辑分离,提升代码复用性和可维护性,本文给大家介绍JavaAOP面向切面编程的概... 目录一、AOP 是什么？二、AOP 的核心概念与实现方式核心概念实现方式三、Spring AOP 的关
阅读更多...
详解SpringBoot+Ehcache使用示例

详解SpringBoot+Ehcache使用示例

《详解SpringBoot+Ehcache使用示例》本文介绍了SpringBoot中配置Ehcache、自定义get/set方式,并实际使用缓存的过程,文中通过示例代码介绍的非常详细,对大家的学习或者... 目录摘要概念内存与磁盘持久化存储：配置灵活性：编码示例引入依赖：配置ehcache.XML文件：配置
阅读更多...
Java 虚拟线程的创建与使用深度解析

Java 虚拟线程的创建与使用深度解析

《Java虚拟线程的创建与使用深度解析》虚拟线程是Java19中以预览特性形式引入,Java21起正式发布的轻量级线程,本文给大家介绍Java虚拟线程的创建与使用,感兴趣的朋友一起看看吧... 目录一、虚拟线程简介1.1 什么是虚拟线程？1.2 为什么需要虚拟线程？二、虚拟线程与平台线程对比代码对比示例：三
阅读更多...
Python版本信息获取方法详解与实战

Python版本信息获取方法详解与实战

《Python版本信息获取方法详解与实战》在Python开发中,获取Python版本号是调试、兼容性检查和版本控制的重要基础操作,本文详细介绍了如何使用sys和platform模块获取Python的主... 目录1. python版本号获取基础2. 使用sys模块获取版本信息2.1 sys模块概述2.1.1
阅读更多...
Java中的.close()举例详解

Java中的.close()举例详解

《Java中的.close()举例详解》.close()方法只适用于通过window.open()打开的弹出窗口,对于浏览器的主窗口,如果没有得到用户允许是不能关闭的,：本文主要介绍Java中的.... 目录当你遇到以下三种情况时，一定要记得使用 .close()：用法作用举例如何判断代码中的 input
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2