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实战应用

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";
    }
}