MQTT協定之訂閱及釋出（使用paho-mqtt-client或mqttv3實作）

另外一個MQTT釋出訂閱用戶端paho-mqtt-client或mqttv3采用回調的方式實作消息的接收，下面看一下實作：

1.消息接收回調類

[java] 

view plain

 copy

 print

?

1. package cn.smartslim.mqtt.demo.paho;  
2. import org.eclipse.paho.client.mqttv3.MqttCallback;  
3. import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;  
4. import org.eclipse.paho.client.mqttv3.MqttMessage;  
5. import org.eclipse.paho.client.mqttv3.MqttTopic;  
6. /**  
7.  * 釋出消息的回調類  
8.  *   
9.  * 必須實作MqttCallback的接口并實作對應的相關接口方法  
10.  *      ◦CallBack 類将實作 MqttCallBack。每個客戶機辨別都需要一個回調執行個體。在此示例中，構造函數傳遞客戶機辨別以另存為執行個體資料。在回調中，将它用來辨別已經啟動了該回調的哪個執行個體。  
11.  *  ◦必須在回調類中實作三個方法：  
13.  *  public void messageArrived(MqttTopic topic, MqttMessage message)  
14.  *  接收已經預訂的釋出。  
16.  *  public void connectionLost(Throwable cause)  
17.  *  在斷開連接配接時調用。  
19.  *  public void deliveryComplete(MqttDeliveryToken token))  
20.  *      接收到已經釋出的 QoS 1 或 QoS 2 消息的傳遞令牌時調用。  
21.  *  ◦由 MqttClient.connect 激活此回調。  
23.  */    
24. public class PushCallback implements MqttCallback {  
25.     public void connectionLost(Throwable cause) {  
26.         // 連接配接丢失後，一般在這裡面進行重連  
27.         System.out.println("連接配接斷開，可以做重連");  
28.     }  
29.     public void deliveryComplete(MqttDeliveryToken token) {  
30.         // publish後會執行到這裡  
31.         System.out.println("deliveryComplete---------"+ token.isComplete());  
33.     public void messageArrived(MqttTopic topic, MqttMessage message) throws Exception {  
34.         // subscribe後得到的消息會執行到這裡面  
35.         System.out.println("接收消息主題:"+topic.getName());  
36.         System.out.println("接收消息Qos:"+message.getQos());  
37.         System.out.println("接收消息内容:"+new String(message.getPayload()));  
39. }  

2.服務端釋出消息

2. import org.eclipse.paho.client.mqttv3.MqttClient;  
3. import org.eclipse.paho.client.mqttv3.MqttConnectOptions;  
5. import org.eclipse.paho.client.mqttv3.MqttException;  
7. import org.eclipse.paho.client.mqttv3.MqttPersistenceException;  
9. import org.eclipse.paho.client.mqttv3.internal.MemoryPersistence;  
10. public class Server {  
11.     public static final String HOST = "tcp://192.168.36.102:1883";  
12.     public static final String TOPIC = "tokudu/yzq124";  
13.     private static final String clientid ="server";   
14.     private MqttClient client;  
15.     private MqttTopic topic;  
16.     private String userName = "test";  
17.     private String passWord = "test";  
18.     private MqttMessage message;  
19.     public Server() throws MqttException {  
20.          //MemoryPersistence設定clientid的儲存形式，預設為以記憶體儲存  
21.         client = new MqttClient(HOST, clientid, new MemoryPersistence());  
22.         connect();  
24.     private void connect() {  
25.         MqttConnectOptions options = new MqttConnectOptions();  
26.         options.setCleanSession(false);  
27.         options.setUserName(userName);  
28.         options.setPassword(passWord.toCharArray());  
29.         // 設定逾時時間  
30.         options.setConnectionTimeout(10);  
31.         // 設定會話心跳時間  
32.         options.setKeepAliveInterval(20);  
33.         try {  
34.                client.setCallback(new PushCallback());  
35.                client.connect(options);  
36.                topic = client.getTopic(TOPIC);  
37.         } catch (Exception e) {  
38.                e.printStackTrace();  
39.         }  
41.     public void publish(MqttMessage message) throws MqttPersistenceException, MqttException{  
42.         MqttDeliveryToken token = topic.publish(message);  
43.         token.waitForCompletion();  
44.         System.out.println(token.isComplete()+"========");  
46.     public static void main(String[] args) throws MqttException {  
47.         Server server =  new Server();  
48.         server.message = new MqttMessage();  
49.         server.message.setQos(1);  
50.         server.message.setRetained(true);  
51.         server.message.setPayload("eeeeeaaaaaawwwwww---".getBytes());  
52.          server.publish(server.message);  
53.          System.out.println(server.message.isRetained()+"------ratained狀态");  

3.用戶端接收消息

2. import java.util.concurrent.Executors;  
3. import java.util.concurrent.ScheduledExecutorService;  
4. import java.util.concurrent.TimeUnit;  
8. import org.eclipse.paho.client.mqttv3.MqttSecurityException;  
11. public class Client {  
14.     private static final String clientid = "client";  
16.     private MqttConnectOptions options;  
19.     private ScheduledExecutorService scheduler;  
20.     //重新連結  
21.     public void startReconnect() {  
22.         scheduler = Executors.newSingleThreadScheduledExecutor();  
23.         scheduler.scheduleAtFixedRate(new Runnable() {  
24.             public void run() {  
25.                 if (!client.isConnected()) {  
26.                     try {  
27.                         client.connect(options);  
28.                     } catch (MqttSecurityException e) {  
29.                         e.printStackTrace();  
30.                     } catch (MqttException e) {  
32.                     }  
33.                 }  
34.             }  
35.         }, 0 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);  
37.     private void start() {  
39.             // host為主機名，test為clientid即連接配接MQTT的用戶端ID，一般以用戶端唯一辨別符表示，MemoryPersistence設定clientid的儲存形式，預設為以記憶體儲存  
40.             client = new MqttClient(HOST, clientid, new MemoryPersistence());  
41.             // MQTT的連接配接設定  
42.             options = new MqttConnectOptions();  
43.             // 設定是否清空session,這裡如果設定為false表示伺服器會保留用戶端的連接配接記錄，這裡設定為true表示每次連接配接到伺服器都以新的身份連接配接  
44.             options.setCleanSession(true);  
45.             // 設定連接配接的使用者名  
46.             options.setUserName(userName);  
47.             // 設定連接配接的密碼  
48.             options.setPassword(passWord.toCharArray());  
49.             // 設定逾時時間 機關為秒  
50.             options.setConnectionTimeout(10);  
51.             // 設定會話心跳時間 機關為秒 伺服器會每隔1.5*20秒的時間向用戶端發送個消息判斷用戶端是否線上，但這個方法并沒有重連的機制  
52.             options.setKeepAliveInterval(20);  
53.             // 設定回調  
54.             client.setCallback(new PushCallback());  
55.             MqttTopic topic = client.getTopic(TOPIC);  
56.             //setWill方法，如果項目中需要知道用戶端是否掉線可以調用該方法。設定最終端口的通知消息    
57.             options.setWill(topic, "close".getBytes(), 0, true);  
58.             client.connect(options);  
59.             //訂閱消息  
60.             int[] Qos  = {1};  
61.             String[] topic1 = {TOPIC};  
62.             client.subscribe(topic1, Qos);  
64.             e.printStackTrace();  
67.     public void disconnect() {  
68.          try {  
69.             client.disconnect();  
70.         } catch (MqttException e) {  
74.     public static void main(String[] args) throws MqttException {     
75.         Client client = new Client();  
76.         client.start();