分散式物聯網平臺EMQX-MQTT

阿新 • • 發佈：2022-05-19

1、安裝
2、整合springboot
- - 報錯處理
3、簡易版客戶端
4、裝置模擬
5、二次開發EMQX的HTTP API

1、安裝

docker部署

docker run -d --name mqtt-broker-emqx-zhgsgl -p 1883:1883 -p 8081:8081 -p 8083:8083 -p 8883:8883 -p 8084:8084 -p 18083:18083 emqx/emqx:v4.0.0

http://192.168.2.201:18083/

admin/public

2、整合springboot

整合springboot文件

報錯處理

整合過程中發生已存在的bean定義，需要在配置檔案新增配置

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {

    void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic);

}

main:
  allow-bean-definition-overriding: true

3、簡易版客戶端

參考：https://www.cnblogs.com/zhuangyao/p/12320500.html

// package com.jtsmartway.zhgsgl.iot.platform.mqttbroker;
//
// import org.eclipse.paho.client.mqttv3.*;
// import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
// import org.springframework.context.annotation.Bean;
// import org.springframework.context.annotation.Configuration;
//
// /**
//  * @author JHL
//  * @version 1.0
//  * @date 2022/5/17 18:07
//  * @since : JDK 11
//  */
// @Configuration
// public class TestEmqxClient {
//
//     
//
//     @Bean
//     public MqttClient getMqttClient() {
//         try {
//             // host為主機名，test為clientid即連線MQTT的客戶端ID，一般以客戶端唯一識別符號表示，MemoryPersistence設定clientid的儲存形式，預設為以記憶體儲存
//             MqttClient client = new MqttClient("tcp://192.168.2.201:1883", "test", new MemoryPersistence());
//             // MQTT的連線設定
//             MqttConnectOptions options = new MqttConnectOptions();
//             // 設定是否清空session,這裡如果設定為false表示伺服器會保留客戶端的連線記錄，這裡設定為true表示每次連線到伺服器都以新的身份連線
//             options.setCleanSession(true);
//             // 設定連線的使用者名稱
//             options.setUserName("admin");
//             // 設定連線的密碼
//             options.setPassword("public".toCharArray());
//             // 設定超時時間 單位為秒
//             options.setConnectionTimeout(10);
//             // 設定會話心跳時間 單位為秒 伺服器會每隔1.5*20秒的時間向客戶端傳送個訊息判斷客戶端是否線上，但這個方法並沒有重連的機制
//             options.setKeepAliveInterval(20);
//             // 設定回撥函式
//             client.setCallback(new MqttCallback() {
//                 @Override
//                 public void connectionLost(Throwable cause) {
//                     System.out.println("connectionLost");
//                 }
//                 @Override
//                 public void messageArrived(String topic, MqttMessage message) throws Exception {
//                     System.out.println(message);
//                     // handleMessage(topic,message);
//                 }
//                 @Override
//                 public void deliveryComplete(IMqttDeliveryToken token) {
//                     System.out.println("deliveryComplete---------" + token.isComplete());
//                 }
//             });
//             client.connect(options);
//             //訂閱訊息
//             // System.out.println(topic);
//             // client.subscribe(topic);
//             return client;
//         } catch (Exception e) {
//             e.printStackTrace();
//             return null;
//         }
//     }
// }