Java MQTT实战应用

2025-06-27 11:23:43 作者：某代码

本文详解MQTT协议,涵盖其发布/订阅机制、低功耗高效特性、三种服务质量等级（QoS0/1/2）,以及客户端、代理、主题的核心概念,最后提供Linux部署教程、SpringBoot整合方法与MQTTX工具下载链接,对java mqtt相关知识感兴趣的朋友一起看看吧

一、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. 客户端认证

4. 创建用户

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;
    }
    // 在客户端连接断开时触发
    @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("deliveryComplete---------" + 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 == mqttTopic) {
            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("发送成功：{}", i);
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "SUCCESS";
    }
}

六、MQTTX官网地址

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

到此这篇关于Java MQTT实战应用的文章就介绍到这了,更多相关Java MQTT内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！