MQTT的學習研究（五） MQTT moquette 的 Blocking API 發佈消息服務端使用

  參看官方文檔：

http://publib.boulder.ibm.com/infocenter/wmqv7/v7r0/index.jsp?topic=/com.ibm.mq.amqtat.doc/tt00000_.htm

 *  Java 爲 MQ Telemetry Transport 建立異步發佈程序
 *在此任務中，您將遵循教程來修改第一個發佈程序。經過修改，
 *使應用程序可以發送發佈而不等待傳遞確認信息。傳遞確認
 *信息由您建立的回調類來接收。
 *
 *
 *
 *4.使客戶機斷開鏈接
 *  a.除去其中包含 token.waitForCompletion 表達式的語句。 主線程將繼續執行，而不等待傳遞發佈。
 *  b.測試客戶機是否已斷開鏈接。 將錯誤返回到 MqttCallback 中的 lostConnection 方法以後，MQTT 客戶機將斷開鏈接，客戶機應用程序也可能斷開鏈接。測試是否有打開的鏈接。
 *  c.使用常量 Example.quiesceTimeout 來設置使客戶機停頓的最長時間。
 *  if (client.isConnected())
 *      client.disconnect(Example.quiesceTimeout);
 *當知足下面三種狀況的組合形式時，客戶機就完成了： 
 *  a.已經對在此會話中（若是從新啓動了會話，則是在先前會話中）已發佈的全部消息調用了回調。
 *  b.消息未完成，然而停頓時間間隔已到期。缺省狀況下，停頓時間間隔爲 30 秒。經過將要等待的毫秒數做爲 client.disconnect 的一個參數來傳遞，便可更改停頓超時。
 *  c.在發佈了某些消息並由客戶機進行排隊以後，可是在發送這些消息以前調用了 client.disconnect。已排隊的消息還沒有處於「未完成」狀態。若是會話可從新啓動，那麼從新啓動會話時就會從新發送消息。
 *  缺省狀況下，停頓時間間隔爲 30 秒。

 

MQTT的消息發佈代碼：

 

package com.etrip.wsmqtt.server;

import com.ibm.micro.client.mqttv3.MqttClient;
import com.ibm.micro.client.mqttv3.MqttDeliveryToken;
import com.ibm.micro.client.mqttv3.MqttMessage;
import com.ibm.micro.client.mqttv3.MqttTopic;
/**
 * 使用 Java 爲 MQ Telemetry Transport 建立異步發佈程序
 *
 *
 *
 *
 * 消息發佈的類的具體的實現
 *
 * @author longgangbai
 *
 */
public class WSMQTTServerPubAsync {
      public static void main(String[] args) {
            try {
                  //建立MqttClient對象
                  MqttClient client = new MqttClient(WSMQTTServerCommon.TCPAddress, WSMQTTServerCommon.clientId);

                  //建立MQTT相關的主題
                  MqttTopic topic = client.getTopic(WSMQTTServerCommon.topicString);

                  //建立MQTT的消息體
                  MqttMessage message = new MqttMessage();
                  //設置消息傳輸的類型
                  message.setQos(2);

                  //設置是否在服務器中保存消息體
                  message.setRetained(false);

                  //設置消息的內容
                  message.setPayload(WSMQTTServerCommon.publication.getBytes());

                  //建立一個MQTT的回調類
                  WSMQTTServerCallBack callback = new WSMQTTServerCallBack(WSMQTTServerCommon.clientId);

                  //MqttClient綁定
                  client.setCallback(callback);

                  //MqttClient鏈接
                  client.connect();

                  System.out.println("Publishing \"" + message.toString()
                      + "\" on topic \"" + topic.getName() + "\" with QoS = "
                      + message.getQos());
                  System.out.println("For client instance \"" + client.getClientId()
                      + "\" on address " + client.getServerURI() + "\"");

                  //發送消息並獲取回執
                  MqttDeliveryToken token = topic.publish(message);

                  System.out.println("With delivery token \"" + token.hashCode()
                      + " delivered: " + token.isComplete());
                  Thread.sleep(100000000000000l);

                  //關閉鏈接
                  if (client.isConnected())
                      client.disconnect(WSMQTTServerCommon.quiesceTimeout);
                  System.out.println("Disconnected: delivery token \"" + token.hashCode()
                      + "\" received: " + token.isComplete());
            } catch (Exception e) {
              e.printStackTrace();
            }
      }
}

 

 

 

MQTT消息發佈回調代碼：

package com.etrip.wsmqtt.server;

import com.ibm.micro.client.mqttv3.*;
/**
 * 發佈消息的回調類
 *
 * 必須實現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 激活此回調。
 *
 * @author longgangbai
 */
public class WSMQTTServerCallBack implements MqttCallback {
      private String instanceData = "";
      public WSMQTTServerCallBack(String instance) {
        instanceData = instance;
      }
      /**
       * 接收到消息的回調的方法
       */
      public void messageArrived(MqttTopic topic, MqttMessage message) {
        try {
          System.out.println("Message arrived: \"" + message.toString()
              + "\" on topic \"" + topic.toString() + "\" for instance \""
              + instanceData + "\"");
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
      /**
       * 消息鏈接丟失
       */
      public void connectionLost(Throwable cause) {
        System.out.println("Connection lost on instance \"" + instanceData
            + "\" with cause \"" + cause.getMessage() + "\" Reason code "
            + ((MqttException)cause).getReasonCode() + "\" Cause \""
            + ((MqttException)cause).getCause() +  "\"");
        cause.printStackTrace();
      }
      /**
       *
       */
      public void deliveryComplete(MqttDeliveryToken token) {
        try {
          System.out.println("Delivery token \"" + token.hashCode()
              + "\" received by instance \"" + instanceData + "\"");
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
}

 

常量類：

package com.etrip.wsmqtt.server;

import java.util.UUID;
/**
 *
 * 消息發佈消息的常量字段
 *
 * @author longgangbai
 */
public final class WSMQTTServerCommon {
  //發佈broker的ip和端口
  public static final String  TCPAddress =System.getProperty("TCPAddress", "tcp://192.168.208.46:1883");
  //客戶端的Id
  public static String clientId =String.format("%-23.23s",  System.getProperty("clientId", (UUID.randomUUID().toString())).trim()).replace('-', '_');
  //發佈消息的主題
  public static final String topicString = System.getProperty("topicString", "china/beijing");
  //發佈的消息
  public static final String publication =System.getProperty("publication", "Hello World " + String.format("%tc", System.currentTimeMillis()));
  //超時時間
  public static final int quiesceTimeout = Integer.parseInt(System.getProperty("timeout", "10000"));

  public static final int  sleepTimeout = Integer.parseInt(System.getProperty("timeout", "10000"));

  public static final boolean cleanSession =Boolean.parseBoolean(System.getProperty("cleanSession", "false"));

  public static final int QoS =Integer.parseInt(System.getProperty("QoS", "1"));

  public static final boolean retained =Boolean.parseBoolean(System.getProperty("retained", "false"));
}