MQTT的學習之Mosquitto發布-訂閱(2)
阿新 • • 發佈:2017-07-31
creat 訂閱模式 pub 測試 方法 ssa clientm art ble
在《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發送的數據
MQTT的學習之Mosquitto發布-訂閱(2)