1. 程式人生 > >MQTT的學習之Mosquitto釋出-訂閱(2)

MQTT的學習之Mosquitto釋出-訂閱(2)

在《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; private
MqttConnectOptions 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傳送的資料