MQTT的學習之Mosquitto釋出-訂閱（2）
阿新 • • 發佈：2019-01-09
在《MQTT的學習之Mosquitto安裝&使用（1）》一文末尾，我已經模擬了釋出-訂閱模式，只是那時在伺服器直接模擬的，並不是java程式碼模擬的。下面貼出Java程式碼
1、首先引入依賴包：
<!-- mosquitto依賴 --> <dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>1.1.0</version> </dependency>
2、程式碼
ClientMQTT：
package org.mqtt.model; 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.MqttTopic;import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; public class ClientMQTT { public static final String HOST = "tcp://192.168.238.133:1883"; public static final String TOPIC = "sensor"; private static final String clientid = "client11"; private MqttClient client; privateMqttConnectOptions options; private String userName = "admin"; private String passWord = "password"; private void start() { try { // host為主機名，clientid即連線MQTT的客戶端ID，一般以唯一識別符號表示，MemoryPersistence設定clientid的儲存形式，預設為以記憶體儲存 client = new MqttClient(HOST, clientid, new MemoryPersistence()); // MQTT的連線設定 options = new MqttConnectOptions(); // 設定是否清空session,這裡如果設定為false表示伺服器會保留客戶端的連線記錄，這裡設定為true表示每次連線到伺服器都以新的身份連線 options.setCleanSession(true); // 設定連線的使用者名稱 options.setUserName(userName); // 設定連線的密碼 options.setPassword(passWord.toCharArray()); // 設定超時時間 單位為秒 options.setConnectionTimeout(10); // 設定會話心跳時間 單位為秒 伺服器會每隔1.5*20秒的時間向客戶端傳送個訊息判斷客戶端是否線上，但這個方法並沒有重連的機制 options.setKeepAliveInterval(20); // 設定回撥 client.setCallback(new PushCallback()); MqttTopic topic = client.getTopic(TOPIC); //setWill方法，如果專案中需要知道客戶端是否掉線可以呼叫該方法。設定最終埠的通知訊息 options.setWill(topic, "close".getBytes(), 2, true); client.connect(options); //訂閱訊息 int[] Qos = {1}; String[] topic1 = {TOPIC}; client.subscribe(topic1, Qos); } catch (Exception e) { e.printStackTrace(); } } public static void main(String[] args) throws MqttException { ClientMQTT client = new ClientMQTT(); client.start(); } }
PushCallback：
package org.mqtt.model; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttMessage; /** * 釋出訊息的回撥類 * * 必須實現MqttCallback的介面並實現對應的相關介面方法 ◦CallBack 類將實現 * MqttCallBack。每個客戶機標識都需要一個回撥例項。在此示例中 * ，建構函式傳遞客戶機標識以另存為例項資料。在回撥中，將它用來標識已經啟動了該回調的哪個例項。 ◦必須在回撥類中實現三個方法： * * public void messageArrived(MqttTopic topic, MqttMessage message) 接收已經預訂的釋出。 * * public void connectionLost(Throwable cause) 在斷開連線時呼叫。 * * public void deliveryComplete(MqttDeliveryToken token)) 接收到已經發布的 QoS 1 或 QoS 2 * 訊息的傳遞令牌時呼叫。 ◦由 MqttClient.connect 啟用此回撥。 * */ public class PushCallback implements MqttCallback { public void connectionLost(Throwable cause) { // 連線丟失後，一般在這裡面進行重連 System.out.println("連線斷開，可以做重連"); } @Override public void deliveryComplete(IMqttDeliveryToken arg0) { // publish後會執行到這裡 System.out.println("deliveryComplete---------" + arg0.isComplete()); } @Override public void messageArrived(String arg0, MqttMessage arg1) throws Exception { System.out.println("接收訊息主題:" + arg0); System.out.println("接收訊息Qos:" + arg1.getQos()); System.out.println("接收訊息內容:" + new String(arg1.getPayload())); } }
ServerMQTT：
package org.mqtt.model; /** * Created by Administrator on 17-2-10. */ import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.MqttPersistenceException; import org.eclipse.paho.client.mqttv3.MqttTopic; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; public class ServerMQTT { //tcp://MQTT安裝的伺服器地址:MQTT定義的埠號 public static final String HOST = "tcp://192.168.238.133:1883"; //public static final String HOST = "tcp://10.10.10.201:1883"; //定義一個主題 public static final String TOPIC = "sensor"; //定義MQTT的ID，可以在MQTT服務配置中指定 private static final String clientid = "server11"; private MqttClient client; private MqttTopic topic11; private String userName = "mosquitto"; private String passWord = ""; private MqttMessage message; /** * 建構函式 * @throws MqttException */ public ServerMQTT() throws MqttException { // MemoryPersistence設定clientid的儲存形式，預設為以記憶體儲存 client = new MqttClient(HOST, clientid, new MemoryPersistence()); connect(); } /** * 用來連線伺服器 */ private void connect() { MqttConnectOptions options = new MqttConnectOptions(); options.setCleanSession(false); options.setUserName(userName); options.setPassword(passWord.toCharArray()); // 設定超時時間 options.setConnectionTimeout(10); // 設定會話心跳時間 options.setKeepAliveInterval(20); try { client.setCallback(new PushCallback()); client.connect(options); topic11 = client.getTopic(TOPIC); } catch (Exception e) { e.printStackTrace(); } } /** * * @param topic * @param message * @throws MqttPersistenceException * @throws MqttException */ public void publish(MqttTopic topic , MqttMessage message) throws MqttPersistenceException, MqttException { MqttDeliveryToken token = topic.publish(message); token.waitForCompletion(); System.out.println("message is published completely! " + token.isComplete()); } /** * 啟動入口 * @param args * @throws MqttException */ public static void main(String[] args) throws MqttException { ServerMQTT server = new ServerMQTT(); server.message = new MqttMessage(); server.message.setQos(1); server.message.setRetained(true); // server.message.setPayload("hello,topic11".getBytes()); for(int i = 0; i < 1000000;i++){ server.message.setPayload(("hello,see" + i).getBytes()); server.publish(server.topic11 , server.message); } // server.publish(server.topic11 , server.message); System.out.println(server.message.isRetained() + "------ratained狀態"); } }
經過測試：ClientMQTT可以實時接收到ServerMQTT傳送的資料