Java MQTT实战应用_Java

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