java MQTT簡單的訂閱訊息例項
阿新 • • 發佈：2019-01-06
1、首先需要啟動Apollo服務（可參照上一篇文章http://blog.csdn.net/qq_29954971/article/details/79279419）
2、新建java project專案
服務端（釋出訊息）：
回撥函式：package com.activemq.mqtt; 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; /** * * Title:Server * Description: 伺服器向多個客戶端推送主題，即不同客戶端可向伺服器訂閱相同主題 * */ public class Server { public static final String HOST = "tcp://192.168.18.120:1883"; public static final String TOPIC = "test-topic"; private static final String clientid = "server"; private MqttClient client; private MqttTopic topic; private MqttTopic topic125; private String userName = "admin"; private String passWord = "password"; private MqttMessage message; public Server() 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); topic = client.getTopic(TOPIC); } catch (Exception e) { e.printStackTrace(); } } 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()); } public static void main(String[] args) throws MqttException { Server server = new Server(); server.message = new MqttMessage(); server.message.setQos(2); server.message.setRetained(true); server.message.setPayload("可以了abc1".getBytes()); server.publish(server.topic , server.message); System.out.println(server.message.isRetained() + "------ratained狀態"); } }
客戶端（訂閱訊息）：package com.activemq.mqtt; 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("連線斷開，可以做重連"); } public void deliveryComplete(IMqttDeliveryToken token) { System.out.println("deliveryComplete---------" + token.isComplete()); } public void messageArrived(String topic, MqttMessage message) throws Exception { // subscribe後得到的訊息會執行到這裡面 System.out.println("接收訊息主題 : " + topic); System.out.println("接收訊息Qos : " + message.getQos()); System.out.println("接收訊息內容 : " + new String(message.getPayload())); } }
package com.activemq.mqtt; import java.util.concurrent.ScheduledExecutorService; 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 Client { public static final String HOST = "tcp://192.168.18.120:61613"; public static final String TOPIC = "test-topic"; private static final String clientid = "client124"; private MqttClient client; private MqttConnectOptions options; private String userName = "admin"; private String passWord = "password"; private ScheduledExecutorService scheduler; 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 { Client client = new Client(); client.start(); } }
PS:
多個客戶端可訂閱同一主題，其clientid必不相同。客戶端在訂閱主題的內容後，但是不同時間啟動，都在啟動後接收到各自資訊，這體現出了伺服器的推送功能。同樣的，傳送的主題資訊，可以在伺服器的topic可以看到，訪問路徑是：http://127.0.0.1:61680/
其實，如若服務端和客戶端相互通訊，即客戶端可以訂閱可以釋出，服務端可以訂閱也可以釋出，則可不區分服務端客戶端，兩邊程式碼幾乎一樣。類似，兩個客戶端都在訂閱同一主題，這時由第三個客戶端釋出這一主題的請求，前兩個客戶端同樣可以接受該主題的內容，這時三個客戶端的程式碼幾乎一樣，只是前兩個是訂閱，後一個是釋出。