三、Mosquitto Java 客戶端實現

　　本文的實現是在 << 一、Mosquitto 介紹&安裝>> << 二、 Mosquitto 的使用說明 >> 兩篇文章搭建好 Mosquitto 服務基礎上實現的。如果你還沒有搭建 Mosquitto 服務 請參考我上述兩篇文章進行 Mosquitto 服務的搭建。 

　　Java 實現 Mosquitto 的客戶端主要使用 Eclipse Paho Java Client 提供的 SDK 來實現的。有興趣的可以直接去 Eclipse Paha 官網下載對應的sdk 和使用說明。

　　一、 准備工作

　　　　本本講解項目是Maven項目、如果還有對 Maven 不了解或者不熟悉的同學可以網上去學習下、本文不在這講解 Maven 的使用。

　　　　添加依賴　　　　

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.0.2</version>
</dependency>

　　二、本文實現 Mosquitto 消息發送主要分為三個類

　　1> ClientMQTT  客戶端類

　　2> PushCallback 消息回調類

　　3> ServerMQTT 服務端類

　　四、 下面將直接上對應的 code

　　1>  客戶端　　

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 ClientMQTT {

    public static final String HOST = "tcp://172.16.192.102:1883";
    public static final String TOPIC = "root/topic/123";
    private static final String clientid = "client11";
    private MqttClient client;
    private MqttConnectOptions options;
    private String userName = "admin";
    private String passWord = "admin";

    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 {
        ClientMQTT client = new ClientMQTT();
        client.start();
    }
}

　　2> 消息回調

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()));
    }
}

　　3> 服務端

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: 服務器向多個客戶端推送主題，即不同客戶端可向服務器訂閱相同主題
 * 
 * @author yueli 2017年9月1日下午17:41:10
 */
public class ServerMQTT {

    // tcp://MQTT安裝的服務器地址:MQTT定義的端口號
    public static final String HOST = "tcp://172.16.192.102:1883";
    // 定義一個主題
    public static final String TOPIC = "root/topic/123";
    // 定義MQTT的ID，可以在MQTT服務配置中指定
    private static final String clientid = "server11";

    private MqttClient client;
    private MqttTopic topic11;
    private String userName = "mosquitto";
    private String passWord = "mosquitto";

    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,topic14".getBytes());
        server.publish(server.topic11, server.message);
        System.out.println(server.message.isRetained() + "------ratained狀態");
    }
}

好了、到這 Java 實現 Mosquiito 客戶端基本已經完成、本實列只是一個 demo 如果正式使用還得根據自己的業務做很多開發。謝謝