EMQX的docker部署与使用(mqtt)

小灰灰 2024-03-03 08:23 92阅读 0赞

## 1：先创建一个挂载emqx的目录 ##

## 2：docker拉去emqx ##

docker pull emqx/emqx:latest

### 2-1：先启动一次eqmx，然后停止，删除容器 ###

docker run -d --name emqx --privileged=true -p 1883:1883 -p 8883:8883 -p 8083:8083 -p 8084:8084 -p 8081:8081 -p 18083:18083  emqx/emqx:latest

### 2-2：然后复制eqmx目录 ###

docker cp emqx:/opt/emqx .

>  *  这个命令是使用 Docker 的 docker cp 命令来从一个正在运行的容器中将文件或目录复制到主机上
>  *  这里是引用
>  *  具体解释如下：
>  *  docker cp：Docker 命令，用于在容器和主机之间复制文件或目录。
>  *  emqx：容器名称或容器 ID，指定要复制文件或目录的容器。
>  *  /opt/emqx：容器内的路径，指定要复制的文件或目录在容器内的位置。
>  *  . 主机内的路径，指定要将文件或目录复制到主机上的位置。. 表示当前用户的主目录。
>  *  综合起来，命令 docker cp emqx:/opt/emqx ~ 表示将容器 emqx 中位于 /opt/emqx 路径下的文件或目录复制到当前用户的主目录下。这可以将容器内的文件或目录复制到主机上进行查看、编辑或处理。

### 2-3：然后停止，删除容器 ###

docker stop emqx
    docker rm emqx

### 2-4：再重新启动容器与挂载 ###

docker run -d --name emqx --privileged=true -p 1883:1883 -p 8883:8883 -p 8083:8083 -p 8084:8084 -p 8081:8081 -p 18083:18083 -v /root/emqx:/opt/emqx emqx/emqx:latest

### 3：登录EMQX的dashboard界面：输入http://127.0.0.1:18083，默认账号:admin，密码:public ###

![在这里插入图片描述][0059973b63864199ae43560dad6c3656.png]

### 3-1：EMQX的网页客户端工具http://www.emqx.io/online-mqtt-client ###

![在这里插入图片描述][2f06bf99c5f54dac8c7ba2fc2d0ffb71.png]

### 3-2：添加客户端连接EMQX的认证方式（账号或者客户id，默认是随意账号或者客户id都可以连接） ###

![在这里插入图片描述][b232640a2349433f978f7c45e1504395.png]  
![在这里插入图片描述][c2fbe803689a45169d5066b42c9b7c14.png]  
![在这里插入图片描述][47a57dca9b7443f3be4421f75539ec71.png]  
![在这里插入图片描述][fb16209962ad43d397e82d717d06a231.png]  
![在这里插入图片描述][801114c7658e4a9db530d332e16d9b83.png]  
![在这里插入图片描述][7b88c96a6e3c4f2493d05aa78373dec6.png]

### 3-3：输入账号密码保存后，客户端连接就需要使用存在的账号和密码才能连接 ###

![在这里插入图片描述][4e22fcf98941497bbae0eb3e1509717d.png]

### 3-4：客户端连接就需要使用存在的账号和密码才能连接 ###

![在这里插入图片描述][9258b9c21339426b91974a973fed5f2d.png]

## 4：桥接模式，可以转发主题的内容到http或者mqtt，如下转发http配置 ##

![在这里插入图片描述][180c6c3a5b094a3a95f4370b057c3910.png]  
![在这里插入图片描述][da0d4d6ab36940dc9db3109f2cf073b1.png]  
![在这里插入图片描述][e27dc800ab2948a8a21e2f31b107f163.png]  
![在这里插入图片描述][50b4c83d618e4e5fb8dacaf8c13c4589.png]  
![在这里插入图片描述][a169a5c56cc347b7b2a1d0fd217533af.png]

## 4-1：使用客户发送消息 ##

!\[(https://img-blog.csdnimg.cn/ddcc72a2feae40d1b451b51385431fd1.png)  
![在这里插入图片描述][d1c3cf23666a49f197df5e08ebb23c54.png]

## 4-2：http服务的服务端与接收到的数据 ##

![在这里插入图片描述][313d43de1d3448639cd9e1a7776ae716.png]

## 4-2：http服务的只接受mqtt内容发的信息，不需要额外信息，设置payload，修改如下 ##

![在这里插入图片描述][9059d47594164770aa4f8838446b0ce3.png]

## 5：EMQX问题分析![在这里插入图片描述][cf334b527f504b69a7b68ee7055cfe18.png] ##

## 6：Java代码的引入使用,pom导入依赖 ##

<dependencies>
       <dependency>
           <groupId>org.eclipse.paho</groupId>
           <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
           <version>1.2.5</version>
       </dependency>
    </dependencies>

## 6-1：然后创建 MQTT 客户端并连接 ##

MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
    MqttConnectOptions options = new MqttConnectOptions();
    options.setUserName(username);
    options.setPassword(password.toCharArray());
    client.connect(options);

>  *  **MqttClient: 同步调用客户端，使用阻塞方法通信**
>  *  **MqttClientPersistence: 代表一个持久的数据存储，用于在传输过程中存储出站和入站的信息，使其能够传递到指定的 QoS**
>  *  **MqttConnectOptions: 连接选项，用于指定连接的参数，下面列举一些常见的方法**
>  *  **setUserName: 设置用户名**
>  *  **setPassword: 设置密码**
>  *  **setCleanSession: 设置是否清除会话**
>  *  **setKeepAliveInterval: 设置心跳间隔**
>  *  **setConnectionTimeout: 设置连接超时时间**
>  *  **setAutomaticReconnect: 设置是否自动重连**

## 6-2：发布 MQTT 消息 ##

package io.emqx.mqtt;
    
    import org.eclipse.paho.client.mqttv3.MqttClient;
    import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    import org.eclipse.paho.client.mqttv3.MqttException;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
    
    public class PublishSample {
        
    
       public static void main(String[] args) {
        
    
           String broker = "tcp://192.168.11.50:1883";
           //String broker = "ws://192.168.11.50:8083/mqtt";//看emqt开启了什么服务
           String topic = "mqtt/test";
           String username = "emqx";
           String password = "public";
           String clientid = "publish_client";
           String content = "Hello MQTT";
           int qos = 0;
    
           try {
        
               MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
               // 连接参数
               MqttConnectOptions options = new MqttConnectOptions();
               // 设置用户名和密码
               options.setUserName(username);
               options.setPassword(password.toCharArray());
               options.setConnectionTimeout(60);
          options.setKeepAliveInterval(60);
               // 连接
               client.connect(options);
               // 创建消息并设置 QoS
               MqttMessage message = new MqttMessage(content.getBytes());
               message.setQos(qos);
               // 发布消息
               client.publish(topic, message);
               System.out.println("Message published");
               System.out.println("topic: " + topic);
               System.out.println("message content: " + content);
               // 关闭连接
               client.disconnect();
               // 关闭客户端
               client.close();
          } catch (MqttException e) {
        
               throw new RuntimeException(e);
          }
      }
    }

## 6-3：订阅 MQTT 主题 ##

package io.emqx.mqtt;
    
    import org.eclipse.paho.client.mqttv3.*;
    import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
    
    public class SubscribeSample {
        
       public static void main(String[] args) {
        
           String broker = "tcp://broker.emqx.io:1883";
           String topic = "mqtt/test";
           String username = "emqx";
           String password = "public";
           String clientid = "subscribe_client";
           int qos = 0;
    
           try {
        
               MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
               // 连接参数
               MqttConnectOptions options = new MqttConnectOptions();
               options.setUserName(username);
               options.setPassword(password.toCharArray());
               options.setConnectionTimeout(60);
          options.setKeepAliveInterval(60);
               // 设置回调
               client.setCallback(new MqttCallback() {
        
    
                   public void connectionLost(Throwable cause) {
        
                       System.out.println("connectionLost: " + cause.getMessage());
                  }
    
                   public void messageArrived(String topic, MqttMessage message) {
        
                       System.out.println("topic: " + topic);
                       System.out.println("Qos: " + message.getQos());
                       System.out.println("message content: " + new String(message.getPayload()));
    
                  }
    
                   public void deliveryComplete(IMqttDeliveryToken token) {
        
                       System.out.println("deliveryComplete---------" + token.isComplete());
                  }
    
              });
               client.connect(options);
               client.subscribe(topic, qos);
          } catch (Exception e) {
        
               e.printStackTrace();
          }
      }
    }

>  *  **MqttCallback 说明：**
>  *  **connectionLost(Throwable cause): 连接丢失时被调用**
>  *  **messageArrived(String topic, MqttMessage message): 接收到消息时被调用**
>  *  **deliveryComplete(IMqttDeliveryToken token): 消息发送完成时被调用**

## 6-4：MqttCallback单独做一个接口修改如下 ##

package com.example.poi.utils;
    
    import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    import org.eclipse.paho.client.mqttv3.MqttCallback;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    
    /**
     * @Author xu
     * @create 2023/9/14 11
     */
    public class SampleCallback implements MqttCallback {
        
        public void connectionLost(Throwable cause) {
        
            System.out.println("connection lost：" + cause.getMessage());
        }
    
        public void messageArrived(String topic, MqttMessage message) {
        
            System.out.println("Received message: \n  topic：" + topic + "\n  Qos：" + message.getQos() + "\n  payload：" + new String(message.getPayload()));
        }
    
        public void deliveryComplete(IMqttDeliveryToken token) {
        
            System.out.println("deliveryComplete");
        }
    }

## 6-5：mqtt回调就可以是使用如下方式 ##

client.setCallback(new SampleCallback());

## 7：emqx通讯，如果消费端断开了一天，但是发送端一直发送消息到emqx，怎么保证当消费端正常后，可以消费断开期间的全部消息 ##

*  **EMQ X 是支持 MQTT 消息队列功能的消息中间件，因此可以通过 EMQ X 实现在消费端断开期间保留消息，并在消费端重新连接后将离线期间的消息推送给消费端。  
    具体实现方法如下：**

## 7-1：在消费端连接 EMQ X 时，设置 clean\_session 参数为 false，这样消费端和 EMQ X 之间的连接不会在消费端断开时自动清除 ##

MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
    // 连接参数
    MqttConnectOptions options = new MqttConnectOptions();
    options.setUserName(username);
    options.setPassword(password.toCharArray());
    options.setConnectionTimeout(60);
    options.setKeepAliveInterval(60);
    /**消费端和 EMQ X 之间的连接不会在消费端断开时自动清除,当断开后，只要会话过期时间到了，才会消失，可以保证只要有会话在，连上后就可以消费断开之间的消息，如果不想消费断开之间的消息就重新生成一个客户端id,即可解决问题*/
    options.setCleanSession(false);

*  ## **注意EQMX的开启cleanSession会话为false的时候，默认的会话过期时间为2小时，可以通过修改EMQX配置目录opt/emqx/emqx.conf里添加会话过期时间，单位为毫秒，如下设置为1个小时** ##

mqtt.session_expiry_interval =3600000

## 7-2：订阅主题时，使用 QoS 级别为 1 或 2，这样如果发送端在消费端断开期间发布消息，它们将被存储在 EMQ X 中 ##

## 7-3：在 EMQ X 中，开启 Retained Message 功能，可以确保消费端在订阅时能够接收到最新的消息，而不是在消费端重新连接后接收到旧的消息 ##

String content="阳光明媚";
    MqttMessage message = new MqttMessage(content.getBytes());
    message. setQos(qos);
    /**保留消息，最新一条记录一定可以被消费，只要有客户端连接都可以消费到最新的消息*/
    message.setRetained(true);
    client.publish(topic, message);

*  **如果发送方发送了很多消息，但消费端断开时间很长，可能会导致 EMQ X 中消息数量过多，建议在 EMQ X 中设置合适的消息过期时间或清理策略，以确保系统性能**

## 8：客户端断开怎么设置自动重连, options.setAutomaticReconnect(true) ##

*  **在设置 options.setAutomaticReconnect(true) 之前，已经调用了 options.setCleanSession(true) 方法，并且客户端与代理断开连接时，代理会清除客户端的会话信息。在这种情况下，自动重新连接可能无法生效**
 *  **要解决这个问题，可以尝试将 options.setCleanSession(true) 改为 options.setCleanSession(false)，以保留客户端的会话信息，从而使得自动重新连接有效**

options.setCleanSession(false);
    options.setAutomaticReconnect(true);

*  **回调类修改，可以观察重连状态**

package com.example.poi.utils;
    
    import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    import org.eclipse.paho.client.mqttv3.MqttCallback;
    import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    
    /**
     * @Author xu
     * @create 2023/9/14 11
     */
    public class SampleCallback implements MqttCallbackExtended {
        
    
        public void connectionLost(Throwable cause) {
        
            System.out.println("connection lost：" + cause.getMessage());
        }
    
        public void messageArrived(String topic, MqttMessage message) {
        
            System.out.println("Received message: \n  topic：" + topic + "\n  Qos：" + message.getQos() + "\n  payload：" + new String(message.getPayload()));
        }
    
        public void deliveryComplete(IMqttDeliveryToken token) {
        
            System.out.println("deliveryComplete");
        }
    
    
        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
        
            System.out.println(serverURI);
            System.out.println(reconnect);
        }
    }

[0059973b63864199ae43560dad6c3656.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/2a3976ac867a476ea77e7abaf5135f10.png
[2f06bf99c5f54dac8c7ba2fc2d0ffb71.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/7d2226090ee34f1f8815ffe7982e50c9.png
[b232640a2349433f978f7c45e1504395.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/d28b4c15dc0c4d8cb7133456a5bfbb09.png
[c2fbe803689a45169d5066b42c9b7c14.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/6007ef33ae7e4695b6b0d5fa66fc6c54.png
[47a57dca9b7443f3be4421f75539ec71.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/65198c9918a54b75b17d3db11b6518d6.png
[fb16209962ad43d397e82d717d06a231.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/886bcb51315840bcbd3b04b35f38eb06.png
[801114c7658e4a9db530d332e16d9b83.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/2a91b766f4ac4dd5995daec1204f4773.png
[7b88c96a6e3c4f2493d05aa78373dec6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/60d52f0fbf7a4b75b59dfebc79cddeb8.png
[4e22fcf98941497bbae0eb3e1509717d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/2dcf53dfa31f4f83ac7c86863ba53f55.png
[9258b9c21339426b91974a973fed5f2d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/020f118af5d84928b27b1359f128b61b.png
[180c6c3a5b094a3a95f4370b057c3910.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/ed242b190d1541c9b0cef3dcd551a562.png
[da0d4d6ab36940dc9db3109f2cf073b1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/41a15e6beb2b4f02b32d80e59266fdca.png
[e27dc800ab2948a8a21e2f31b107f163.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/141a85c41457484498244ad7f8a083e7.png
[50b4c83d618e4e5fb8dacaf8c13c4589.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/c021167f7f664a659b34d688701f64d5.png
[a169a5c56cc347b7b2a1d0fd217533af.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/5965f1bdf57d4d92b97c3019c1909aca.png
[d1c3cf23666a49f197df5e08ebb23c54.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/63e9400782d544aa846ee2b738683bf6.png
[313d43de1d3448639cd9e1a7776ae716.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/8950ce7e56f54dcda40a8dab862f8036.png
[9059d47594164770aa4f8838446b0ce3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/5239d983bcfa499ebb0e8245f9ec3e6e.png
[cf334b527f504b69a7b68ee7055cfe18.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/5ce50736116349c984af2527c97e7ab9.png