物聯網mqtt協議是可以發布和訂閱通過java就可以實作
話不多說，mqtt，加載pom.xml檔案格式
1 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 2 <modelVersion>4.0.0</modelVersion> 3 <groupId>com.mqtt</groupId> 4 <artifactId>MQTTmosquitto</artifactId> 5 <version>0.0.1-SNAPSHOT</version> 6 <dependencies> 7 <!-- MQTT依賴 --> 8 <dependency> 9 <groupId>org.eclipse.paho</groupId> 10 <artifactId>org.eclipse.paho.client.mqttv3</artifactId> 11 <version>1.0.2</version> 12 </dependency> 13 <!-- JDBC --> 14 <dependency> 15 <groupId>mysql</groupId> 16 <artifactId>mysql-connector-java</artifactId> 17 <version>5.1.39</version> 18 </dependency> 19 20 <!-- 阿里巴巴JSON --> 21 <dependency> 22 <groupId>com.alibaba</groupId> 23 <artifactId>fastjson</artifactId> 24 <version>1.2.28</version> 25 </dependency> 26 27 28 </dependencies> 29 30 <build> 31 <plugins> 32 <plugin> 33 <artifactId>maven-compiler-plugin</artifactId> 34 <version>2.3.2</version> 35 <configuration> 36 <source>1.8</source> 37 <target>1.8</target> 38 </configuration> 39 </plugin> 40 <plugin> 41 <artifactId> maven-assembly-plugin </artifactId> 42 <configuration> 43 <descriptorRefs> 44 <descriptorRef>MQTT</descriptorRef> 45 </descriptorRefs> 46 <archive> 47 <manifest> 48 <mainClass>com.mqtt.ServerMQTT</mainClass> 49 </manifest> 50 </archive> 51 </configuration> 52 <executions> 53 <execution> 54 <id>make-assembly</id> 55 <phase>package</phase> 56 <goals> 57 <goal>single</goal> 58 </goals> 59 </execution> 60 </executions> 61 </plugin> 62 </plugins> 63 </build> 64 65 </project>
這塊是發布資訊代碼
1 package com.mqtt; 2 3 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; 4 import org.eclipse.paho.client.mqttv3.MqttCallback; 5 import org.eclipse.paho.client.mqttv3.MqttClient; 6 import org.eclipse.paho.client.mqttv3.MqttConnectOptions; 7 import org.eclipse.paho.client.mqttv3.MqttDeliveryToken; 8 import org.eclipse.paho.client.mqttv3.MqttException; 9 import org.eclipse.paho.client.mqttv3.MqttMessage; 10 import org.eclipse.paho.client.mqttv3.MqttPersistenceException; 11 import org.eclipse.paho.client.mqttv3.MqttTopic; 12 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; 13 14 public class ServerMQTT { 15 16 public static void main(String[] args) throws MqttException { 17 ServerMQTT server = new ServerMQTT(); 18 19 server.message = new MqttMessage(); 20 server.message.setQos(2); 21 server.message.setRetained(true); 22 //編輯訊息 23 server.message.setPayload("你的肉".getBytes()); 24 server.publish(server.topic , server.message); 25 System.out.println(server.message.isRetained() + "------ratained狀態"); 26 } 27 28 //MQTT安裝的服務器地址和埠號 29 public static final String HOST = "tcp://localhost:1883"; 30 //定義一個主題 31 public static final String TOPIC = "test"; 32 //定義MQTT的ID，可以在MQTT服務配置中指定 33 private static final String clientid = "client-1"; 34 35 private MqttClient client; 36 private MqttTopic topic;
//用戶和密碼 37 private String userName = "zhny"; 38 private String passWord = "zhny2020"; 39 private MqttMessage message; 40 41 /** 42 * g建構式 43 */ 44 public ServerMQTT() throws MqttException { 45 // MemoryPersistence設定clientid的保存形式，默認為以記憶體保存 46 client = new MqttClient(HOST, clientid, new MemoryPersistence()); 47 connect(); 48 } 49 50 /** 51 * l連接服務器 52 */ 53 private void connect() { 54 MqttConnectOptions options = new MqttConnectOptions(); 55 options.setCleanSession(false); 56 options.setUserName(userName); 57 options.setPassword(passWord.toCharArray()); 58 // 設定超時時間 59 options.setConnectionTimeout(10); 60 // 設定會話心跳時間 61 options.setKeepAliveInterval(20); 62 try { 63 client.setCallback(new MqttCallback() { 64 public void connectionLost(Throwable cause) { 65 // 連接丟失后，一般在這里面進行重連 66 System.out.println("連接斷開……（可以做重連）"); 67 } 68 69 public void deliveryComplete(IMqttDeliveryToken token) { 70 System.out.println("deliveryComplete---------" + token.isComplete()); 71 } 72 73 public void messageArrived(String topic, MqttMessage message) throws Exception { 74 // subscribe后得到的訊息會執行到這里面 75 System.out.println("接收訊息主題:" + topic + " 接收訊息Qos:" + message.getQos() + "接收訊息內容:" + new String(message.getPayload())); 76 } 77 }); 78 client.connect(options); 79 80 topic = client.getTopic(TOPIC); 81 } catch (Exception e) { 82 e.printStackTrace(); 83 } 84 } 85 86 /** 87 * t推送訊息 88 */ 89 public void publish(MqttTopic topic , MqttMessage message) throws MqttPersistenceException, MqttException { 90 MqttDeliveryToken token = topic.publish(message); 91 token.waitForCompletion(); 92 System.out.println("測驗成功為true失敗為false " + token.isComplete()); 93 } 94 95 96 97 }
訂閱接收代碼
1 package com.mqtt; 2 3 import org.eclipse.paho.client.mqttv3.MqttClient; 4 import org.eclipse.paho.client.mqttv3.MqttConnectOptions; 5 import org.eclipse.paho.client.mqttv3.MqttException; 6 import org.eclipse.paho.client.mqttv3.MqttTopic; 7 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; 8 9 /** 10 * MQTT工具類操作 11 * 12 */ 13 public class MQTTConnect { 14 15 16 public static void main(String[] args) throws MqttException { 17 MQTTConnect client = new MQTTConnect(); 18 client.start(); 19 } 20 21 22 23 24 25 //MQTT安裝的服務器地址和埠號（本機的ip） 26 public static final String HOST = "tcp://localhost:1883"; 27 //定義一個主題 28 public static final String TOPIC = "test"; 29 //定義MQTT的ID，可以在MQTT服務配置中指定 30 private static final String clientid = "client-2"; 31 private MqttClient client; 32 private MqttConnectOptions options;
//用戶和密碼 33 private String userName = "zhny"; 34 private String passWord = "zhny2020"; 35 36 // private ScheduledExecutorService scheduler; 37 38 public void start() { 39 try { 40 // host為主機名，clientid即連接MQTT的客戶端ID，一般以唯一識別符號表示，MemoryPersistence設定clientid的保存形式，默認為以記憶體保存 41 client = new MqttClient(HOST, clientid, new MemoryPersistence()); 42 // MQTT的連接設定 43 options = new MqttConnectOptions(); 44 // 設定是否清空session,這里如果設定為false表示服務器會保留客戶端的連接記錄，這里設定為true表示每次連接到服務器都以新的身份連接 45 options.setCleanSession(true); 46 // 設定連接的用戶名 47 options.setUserName(userName); 48 // 設定連接的密碼 49 options.setPassword(passWord.toCharArray()); 50 // 設定超時時間 單位為秒 51 options.setConnectionTimeout(10); 52 // 設定會話心跳時間 單位為秒 服務器會每隔1.5*20秒的時間向客戶端發送個訊息判斷客戶端是否在線，但這個方法并沒有重連的機制 53 options.setKeepAliveInterval(20); 54 55 // 設定回呼,client.setCallback就可以呼叫PushCallback類中的messageArrived()方法 56 client.setCallback(new PushCallback()); 57 MqttTopic topic = client.getTopic(TOPIC); 58 59 int qos = 2; 60 61 //setWill方法，如果專案中需要知道客戶端是否掉線可以呼叫該方法，設定最終埠的通知訊息 62 //options.setWill(topic, "This is yizhu...".getBytes(), qos, true); 63 client.connect(options); 64 //訂閱訊息 65 int[] Qos = {qos}; 66 String[] topic1 = {TOPIC}; 67 client.subscribe(topic1, Qos); 68 69 } catch (Exception e) { 70 e.printStackTrace(); 71 } 72 } 73 74 } 75 76 77 78 79 80 81 82
1 package com.mqtt; 2 3 4 5 6 7 import java.text.SimpleDateFormat; 8 9 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; 10 import org.eclipse.paho.client.mqttv3.MqttCallback; 11 import org.eclipse.paho.client.mqttv3.MqttMessage; 12 13 import com.alibaba.fastjson.JSONObject; 14 import com.jdbc.MySQLDemo; 15 16 17 18 /** 19 * 必須實作MqttCallback的介面并實作對應的相關介面方法CallBack 類將實作 MqttCallBack， 20 * 每個客戶機標識都需要一個回呼實體，在此示例中，建構式傳遞客戶機標識以另存為實體資料， 21 * 在回呼中，將它用來標識已經啟動了該回呼的哪個實體， 22 * 必須在回呼類中實作三個方法： 23 * public void messageArrived(MqttTopic topic, MqttMessage message)接收已經預訂的發布， 24 * public void connectionLost(Throwable cause)在斷開連接時呼叫， 25 * public void deliveryComplete(MqttDeliveryToken token)) 26 * 接收到已經發布的 QoS 1 或 QoS 2 訊息的傳遞令牌時呼叫， 27 * 由 MqttClient.connect 激活此回呼， 28 */ 29 public class PushCallback implements MqttCallback{ 30 31 32 public void connectionLost(Throwable cause) { 33 // 連接丟失后，一般在這里面進行重連 34 System.out.println("連接斷開……（可以做重連）"); 35 } 36 37 38 39 public void deliveryComplete(IMqttDeliveryToken token) { 40 System.out.println("deliveryComplete---------" + token.isComplete()); 41 } 42 43 44 public void messageArrived(String topic, MqttMessage message) throws Exception { 45 // subscribe后得到的訊息會執行到這里面 46 System.out.println("接收訊息主題:" + topic + " 接收訊息Qos:" + message.getQos() + "接收訊息內容:" + new String(message.getPayload())); 47
//特別注意一下，messageArrived這個里面不允許呼叫代碼和寫方法，如果呼叫就會斷開連接這個是個坑，然后用我這個try{}catch{}就沒有問題了，就不會再去掉線斷開連接
try { 48 JSONObject jsonObject = JSONObject.parseObject(message.toString()); 49 String equipment=jsonObject.getString("equipment"); 50 double temperature=Double.parseDouble(jsonObject.getString("temperature")); 51 double humidity=Double.parseDouble(jsonObject.getString("humidity")); 52 String datetime=jsonObject.getString("datetime"); 53 SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd hh-mm-ss"); 54 System.out.println(equipment); 55 56 MySQLDemo mysqldemo=new MySQLDemo(); 57 mysqldemo.add(equipment, temperature, humidity, datetime); 58 } catch (Exception e) { 59 e.printStackTrace(); 60 } 61 62 } 63 64 private void MqttPublisherFetch() { 65 // 66 // JSONObject jsonObject = JSONObject.parseObject(messtr); 67 // 68 // jsonObject.put("李鵬", "1111"); 69 // 70 // jsonObject.put("李12", "222"); 71 // 72 // System.out.println(jsonObject.toString()); 73 74 // System.out.println(equipment+"-------"+temperature+"---------------"+humidity+"-----------"+datetime+"-----------豬蹄---"); 75 } 76 77 78 } 79 80
特別注意一下，messageArrived這個里面不允許呼叫代碼和寫方法，如果呼叫就會斷開連接這個是個坑，然后用我這個try{}catch{}就沒有問題了，就不會再去掉線斷開連接
簡單的一個小maven專案集成mqtt還有個jdbc簡單的也給大家粘上去
1 package com.jdbc; 2 3 import java.sql.*; 4 5 import com.alibaba.fastjson.JSONObject; 6 7 public class MySQLDemo { 8 9 // MySQL 8.0 以下版本 - JDBC 驅動名及資料庫 URL 10 static final String JDBC_DRIVER = "com.mysql.jdbc.Driver"; 11 static final String DB_URL = "jdbc:mysql://localhost:3306/zhny?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true"; 12 13 // MySQL 8.0 以上版本 - JDBC 驅動名及資料庫 URL 14 //static final String JDBC_DRIVER = "com.mysql.cj.jdbc.Driver"; 15 //static final String DB_URL = "jdbc:mysql://localhost:3306/RUNOOB?useSSL=false&serverTimezone=UTC"; 16 17 18 // 資料庫的用戶名與密碼，需要根據自己的設定 19 static final String USER = "root"; 20 static final String PASS = "root"; 21 22 23 public void select() { 24 Connection conn = null; 25 Statement stmt = null; 26 try{ 27 // 注冊 JDBC 驅動 28 Class.forName(JDBC_DRIVER); 29 30 // 打開鏈接 31 System.out.println("連接資料庫..."); 32 conn = DriverManager.getConnection(DB_URL,USER,PASS); 33 34 35 // 執行查詢 36 System.out.println(" 實體化Statement物件..."); 37 stmt = conn.createStatement(); 38 String sql; 39 sql = "select * from crop"; 40 ResultSet rs = stmt.executeQuery(sql); 41 // 展開結果集資料庫 42 while(rs.next()){ 43 // 通過欄位檢索 44 int id = rs.getInt("id"); 45 String name = rs.getString("crop_name"); 46 47 // 輸出資料 48 System.out.print("ID: " + id); 49 System.out.print(", 站點名稱: " + name); 50 System.out.print("\n"); 51 } 52 // 完成后關閉 53 rs.close(); 54 stmt.close(); 55 conn.close(); 56 }catch(SQLException se){ 57 // 處理 JDBC 錯誤 58 se.printStackTrace(); 59 }catch(Exception e){ 60 // 處理 Class.forName 錯誤 61 e.printStackTrace(); 62 }finally{ 63 // 關閉資源 64 try{ 65 if(stmt!=null) stmt.close(); 66 }catch(SQLException se2){ 67 }// 什么都不做 68 try{ 69 if(conn!=null) conn.close(); 70 }catch(SQLException se){ 71 se.printStackTrace(); 72 } 73 } 74 System.out.println("Goodbye!"); 75 } 76 77 public void add(String equipment,double temperature,double humidity,String datetime) { 78 79 80 Connection conn = null; 81 Statement stmt = null; 82 try{ 83 // 注冊 JDBC 驅動 84 Class.forName(JDBC_DRIVER); 85 86 // 打開鏈接 87 System.out.println("連接資料庫..."); 88 conn = DriverManager.getConnection(DB_URL,USER,PASS); 89 90 91 // 執行查詢 92 System.out.println(" 實體化Statement物件..."); 93 stmt = conn.createStatement(); 94 String sql = "INSERT INTO sensor (model,device_id,temperature,humidity,date_time,section_id)\r\n" + 95 "VALUES('"+equipment+"','"+temperature+"','"+temperature+"','"+humidity+"','"+datetime+"',12)";//傳感器 96 int rs = stmt.executeUpdate(sql); 97 if (rs>0) { 98 System.out.println("true"+rs); 99 }else { 100 System.out.println("flase"+rs); 101 } 102 103 104 105 stmt.close(); 106 conn.close(); 107 }catch(SQLException se){ 108 // 處理 JDBC 錯誤 109 se.printStackTrace(); 110 }catch(Exception e){ 111 // 處理 Class.forName 錯誤 112 e.printStackTrace(); 113 }finally{ 114 // 關閉資源 115 try{ 116 if(stmt!=null) stmt.close(); 117 }catch(SQLException se2){ 118 }// 什么都不做 119 try{ 120 if(conn!=null) conn.close(); 121 }catch(SQLException se){ 122 se.printStackTrace(); 123 } 124 } 125 System.out.println("Goodbye!"); 126 } 127 }
最后接收結果
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/1054.html
標籤：領域驅動設計