MQTT 消息 發布 訂閱

---

 

當連接向一個mqtt服務器時，clientId必須是唯一的。設置一樣，導致client.setCallback總是走到 connectionLost回調。報connection reset。調查一天才發現是clientid重復導致。

client = new MqttAsyncClient(serverURIString, "client-id");

 

---

 

clientId是用來保存會話信息。

MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);

當服務器將message發布向所有訂閱過的客戶端，就會清除這個message。如果當前客戶端不在線，等它連接時發送。

 

session與client之間的關系是怎樣的？

這樣的，比如你一個板子，作為客戶端，發起mqtt的連接請求connect到mqtt服務器，比如說就是emqtt服務吧，emqtt服務端收到這個板子的連接請求之后，在tcp層上會和板子建立一個tcp的連接，在emqtt內部，會產生一個進程，和這個板子做數據通訊，同時還會產生一個進程，叫session，這個sessoin是專門管理這個板子訂閱的主題，其它板子如果發布了這個板子感興趣的主題的時候，也會發到這個板子對應的這個session里面，如果這個session收到訂閱的主題之后，發現對用的client還活着，就通過這個client把數據經過tcp發到這個板子上，如果發現client已經沒有了，就是說板子和服務端斷掉了，那么session就會把收到的訂閱的主題，先保存在session里面，下次板子連接上了，而且cleansession=false，那么這個session就不會清除，在這次連接時，就會把以前收到的訂閱消息，發給板子，大概就是這個意思。

---

 

參考：

http://www.blogjava.net/yongboy/archive/2014/02/15/409893.html

http://www.cnblogs.com/znlgis/p/4930990.html

http://blog.csdn.net/ljf10010/article/details/51424506

paho客戶端示例

 

https://github.com/eclipse/paho.mqtt.java/tree/master/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test

http://www.eclipse.org/paho/files/javadoc/index.html api文檔

ibm客戶端paho示例：

http://www.programcreek.com/java-api-examples/index.php?source_dir=streamsx.messaging-master/com.ibm.streamsx.messaging/impl/java/src/com/ibm/streamsx/messaging/mqtt/MqttAsyncClientWrapper.java

hivemq的mqtt詳解：http://www.hivemq.com/blog/mqtt-essentials-part-8-retained-messages

 

 http://blog.csdn.net/wuyinxian/article/details/38826259

MQTT Protocol Manual（Apollo中MQTT協議解析）

 

 

 

        官網英文參考： http://activemq.apache.org/apollo/documentation/mqtt-manual.html

 

 MQTT協議

Apollo允許客戶端通過開放的MQTT協議連接。該協議主要是用在資源有限的驅動上，以及網絡不穩定的情況下使用，是一個訂閱、發布模型。這種驅動通常不適用類似http，stomp這類基於文本，或者類似openfire，AMQP等傳統二進制協議。MQTT是一個簡介的二進制協議，適用這類驅動資源受限，而且是不穩定的網絡條件下。

之前的穩定發布版本中，MQTT是作為一個Apollo的一個插件提供的。但是現在，這個插件已經變為開發項目的一部分。MQTT在Apollo中已經不需要其他配置文件或者是第三方插件支持了。

MQTT是一個線路層的協議，任何實現該協議的客戶端都可以連接到Apollo。當然也可以整合其他MQTT兼容的消息代理中。

更多有關MQTT協議內容，參考 the MQTT Specification

 MQTT協議配置

為了開始使用MQTT協議，首先使用MQTT3.1協議的客戶端，連接到Apollo正在監聽端口。Apollo會做協議檢測，而且自動識別MQTT連接，而且將連接作為MQTT協議處理。你不必要為MQTT協議打開一個端口（STomp，Openfire，AMQP等都是自動識別）。如果你一定指定連接的協議，有下面兩種方式：你可以選擇不用協議識別，而是為MQTT指定連接：

<connector id="tcp" bind="tcp://0.0.0.0:61613" protocol="mqtt"/>

或者你可以限制哪種協議可以被自動識別。通過下面的<detece>配置方式：

<connector id="tcp" bind="tcp://0.0.0.0:61613">

  <detect protocols="mqtt openwire" />

</connector>

<detect> 下protocols 對應的參數通過空格來隔開支持的通信協議。如果只支持一種協議，就不要空格，默認情況下對任何協議生效。

如果你想調整MQTT默認設置，在apollo.xml文件中有一個<connector> 元素，通過MQTT參數配置：

<connector id="tcp" bind="tcp://0.0.0.0:61613">

  <mqtt max_message_length="1000" />

</connector>

MQTT元素支持下面幾個參數：

• max_message_length : The size (in bytes) of the largest message that can be sent to the broker. Defaults to 100MB（broker能接受的最大消息量：默認是100M）
• protocol_filters : A filter which can filter frames being sent/received to and from a client. It can modify the frame or even drop it.(一個控制發送和接收，Client的過濾器框架。可以修改，刪除這個框架)
• die_delay : How long after a connection is deemed to be “dead” before the connection actually closes; default: 5000ms（在實際斷開連接之前，會有默認5000ms的時間被認為連接已經dead）

mqtt 配置元素也可以用來控制目的消息頭的解析。下面是支持的參數：

• queue_prefix : a tag used to identify destination types; default: null（用來確認目的地類型）
• path_separator : used to separate segments in a destination name; default: /（用來分割目的地名稱）
• any_child_wildcard : indicate all child-level destinations that match the wildcard; default: +（識別子目錄）
• any_descendant_wildcard : indicate destinations that match the wildcard recursively; default:#（目標地址通配符）
• regex_wildcard_start : pattern used to identify the start of a regex（表示正則表達開始）
• regex_wildcard_end : pattern used to identify the end of a regex（表示正則表達結束）
• part_pattern : allows you to specify a regex that constrains the naming of topics. （你可以指定正則表達規則）default: [ a-zA-Z0-9\_\-\%\~\:\(\)]+

 Client 可用函數庫

Apollo 支持MQTT3.1 協議，下面是可用的Clients:

• Java : mqtt-client, MeQanTT
• C : libmosquitto
• Erlang : erlmqtt, my-mqtt4erl
• .NET : MQTTDotNet, nMQTT
• Perl : net-mqtt-perl, [anyevent-mqtt-perl]https://github.com/beanz/anyevent-mqtt-perl()
• Python : nyamuk
• Ruby : mqtt-ruby, ruby-em-mqtt
• Javascript : Node.js MQTT Client
• Delphi : TMQTTCLient
• Device specific: Arduino, mbed, Nanode, Netduino

如果要找到新支持的Clients ，可以檢索： the MQTT website for its software

在目錄example 目錄下，你可以找到一些例子，實現了與broker之間收發。

 connecting

為了確保broker配置文件的安全，所以只允許一個admin 用戶連接，默認的用戶名和密碼是：admin ,password.

Mqtt 客戶端不能specify 虛擬主機（更多請看：see  the section on Virtual Hosts in the user guide)，以至於默認情況下虛擬主機已經被使用了。通常第一虛擬主機定義在apollo.xml文件中。

 Destination 類型

MQTT協議是訂閱，發布協議，是不允許真正的利用隊列點對點的消息收發。因此Apollo僅允許利用主題，還進行MQTT消息發送。訂閱的概念和持久的主題訂閱 和其他協議提到的有些類似，同時也被MQTT CONNECT 框架的clean session屬性控制。

 Clean Sessions

但一個Client 發送一個連接，這個連接中clean session 被設置為false,那么之前連接中有相同Client_id 的session 將會被重復使用。這就意味着Client斷開了，訂閱依然能收到消息。這就等同與同Apollo建立一個長訂閱。

如果 clean session 設置為true ，那么新session就開始了，其他的session會慢慢消失，刪除。這就是Apollo中定義的普通的主題訂閱。

 Topic Retained Messages

如果消息被發布的同時retain 標記被設置，消息將被主題記住，以至於新的訂閱到達，最近的retain 消息會被發送到訂閱者。比如說：你想發布一個參數，而且你想讓最新的這個參數發布到總是可用的訂閱了這個主題的客戶端上，你就設置在PUBLISH 框架上設置retain 標簽。

注意：retained 消息 不會被設置成retained 在 QoS設置為零的broker 重啟過程中。

Last Will and Testament Message

當Client第一次連接的時候，有一個will 消息和一個更QoS相關的消息會跟你有關。will消息是一個基礎消息，這個基礎消息只有在連接異常或者是掉線的時候才會被發送。一般用在你有一個設備，當他們掉了的時候，你需要知道。所以如果一個醫療Client從broker掉線，will消息將會作為一個鬧鍾主題發送，而且會被系統作為高優先級提醒。

Reliable Messaging

MQTT協議允許Client 發布消息的時候指定Qos參數：

• At Most Once (QoS=0)
• At Least Once (QoS=1)
• Exactly Once (QoS=2)

 最多一次

這個設置時推送消息給Client，可靠性最低的一種。如果設置Qos=0，那broker就不會返回結果碼，告訴你他收到消息了，也不會在失敗后嘗試重發。這有點像不可靠消息，如JMS。

 至少一次

該設置會確保消息會被至少一次推送到Client。如果推送設置為至少推送一次，Apollo會返回一個回調函數，確保代理已經收到消息，而且確保會確保推送該消息。如果Client 將發布了一個Qos=1的消息，如果在指定的時間內沒有收到回復，Client會希望重新發布這個消息。所以可能存在這種情況：代理收到一個需要推送的消息，然后又收到一個消息推送到同一個Client。所以如果傳輸過程中PUBACK丟失，Client會重新發送，而且不會去檢測是否是重發，broker就將消息發送到訂閱主題中。

 恰好一次

該設置是可靠等級最高的。他會確保發布者不僅僅會推送，而且不會像Qos=1 那樣，會被接收兩次。當然這個設置會增加網絡的負載。當一個消息被發布出去的時候，broker會保存該消息的id,而且會利用任何長連接，堅持要把該消息推送給目標地址。如果Client收到PUBREC 標志，那就表明broker已經收到消息了。 這個時候broker會期待Client發送一個PUBREL 來清除session 中消息id，broker如果發送成功就會發送一個PUBCOMP通知Client。

Wildcard Subscriptions

通配用在主題的目標地址中。這能實現一個主題發送到多個用戶，或者多層用戶中。

• / is used to separate names in a path（分割路徑）
• + is used to match any name in a path（通配地址任何字符）
• # is used to recursively match path names（遞歸通配）

比如通配可能這樣來用：

• PRICE/# : Any price for any product on any exchange（任何交易中任何產品的價格）
• PRICE/STOCK/# : Any price for a stock on any exchange（任何交易中的股票價格）
• PRICE/STOCK/NASDAQ/+ : Any stock price on NASDAQ（納斯達克的任何股票價格）
• PRICE/STOCK/+/IBM : Any IBM stock price on any exchange（任何交易中IBM股票價格）

Keep Alive

Apollo只有在Client指定了CONNECT的KeepAlive 值的時候，才會設置保持連接、心跳檢測。如果one Client指定了keepalive,apollo 將會使用1.5*keepalive值。這個在MQTT中有說明。

Destination Name Restrictions

路徑名稱限制了使用（a-z, A-Z, 0-9, _, - %, ~, :, ' ', '(', ')' ，. ）字符，通配符（*）在復雜的分隔符中。而且確保使用utf-8來編譯你的URL。

package com.xxx.mqtt;

import java.net.URI;

import org.eclipse.paho.client.mqttv3.DisconnectedBufferOptions;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
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.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;


public class MyMqttClient implements MqttCallback {

    private static final MemoryPersistence DATA_STORE = new MemoryPersistence();
    private static final String topic = "mytopic";

    private String HOST = "127.0.0.1";
    private int PORT = 1883;
    private String USERNAME = "user";
    private String PASSWORD = "password";
    private String serverURIString = "tcp://" + HOST + ":" + PORT;
    
    String clientId = "client-1";

    MqttAsyncClient client;
    // Tokens
    IMqttToken connectToken;
    IMqttDeliveryToken pubToken;


    public static void main(String[] args) {
        MyMqttClient app = new MyMqttClient();
        app.asyncClient();
        try {
            Thread.sleep(20000);
            app.disconnect();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("end");
    }

    public void blockingClient() {
        
        try {
            MqttClient sampleClient = new MqttClient(serverURIString, clientId);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            connOpts.setUserName(USERNAME);
            connOpts.setPassword(PASSWORD.toCharArray());
            System.out.println("Connecting to broker: " + serverURIString);
            sampleClient.connect(connOpts);
            sampleClient.subscribe("#", 1);
            System.out.println("Connected");
//                System.out.println("Publish message: " + content);
//                MqttMessage message = new MqttMessage(content.getBytes());
//                message.setQos(qos);
            sampleClient.setCallback(this);
//                sampleClient.publish(topic, message);
//                System.out.println("Message published");
            try {
                Thread.sleep(10000000);
                System.out.println("Disconnected");
                sampleClient.disconnect();
            } catch (Exception e) {
                e.printStackTrace();
            }

        } catch (MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("except " + me);
            me.printStackTrace();
        }
    }

    public void asyncClient() {
        info(" MQTT init start.");

        // Tokens
        IMqttToken connectToken;
        IMqttDeliveryToken pubToken;

        // Client Options
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
        options.setAutomaticReconnect(true);

        options.setUserName(USERNAME);
        options.setPassword(PASSWORD.toCharArray());

        try {
            client = new MqttAsyncClient(serverURIString, clientId);

            DisconnectedBufferOptions disconnectedOpts = new DisconnectedBufferOptions();
            disconnectedOpts.setBufferEnabled(true);
            client.setBufferOpts(disconnectedOpts);

            connectToken = client.connect(options);
            connectToken.waitForCompletion();//異步變成了同步。可以用IMqttCallbackListen..在connect時候設置回調。
            boolean isConnected = client.isConnected();
            info("Connection isConnected: " + isConnected);

            if (connectToken.isComplete() && connectToken.getException() == null && client.isConnected()) {
                info("[Connect:] Success: "); //$NON-NLS-1$ //$NON-NLS-2$ 
                client.setCallback(this);

            } else {
                info("[Connect:] faild: "); //$NON-NLS-1$ //$NON-NLS-2$ 
            }

//             MqttTopic topic = client.getTopic(topic); 
//             topic.
            //setWill方法，如果項目中需要知道客戶端是否掉線可以調用該方法。設置最終端口的通知消息    
//                options.setWill(topic, "close".getBytes(), 2, true);  

            IMqttToken subToken = client.subscribe("#", 1);

            subToken.waitForCompletion(1000);

            if (subToken.isComplete()) {
                info("subToken  complete.");
                if (subToken.getException() != null) {
                    info("Topics Subscription failed." + subToken.getException()); //$NON-NLS-1$ 
                }
            } else {
                info("subToken not complete.");
                if (subToken.getException() != null) {
                    info("Topics Subscription failed." + subToken.getException()); //$NON-NLS-1$ 
                }
            }

            info("init end");

        } catch (MqttException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }

//String clientId, String topic, String message
    public void send() {
        String topic;
        String message;
        info("===Send Message start.===");
        message = "Hello, boy.";
        

        boolean isConnected = client.isConnected();
        if (!isConnected) {
            //no need. it will auto reconnect and send.
        }

        // Publish Message
        try {
            pubToken = client.publish(topic, new MqttMessage(message.getBytes()));

            info("Publish attempted: isComplete:" + pubToken.isComplete());

            pubToken.waitForCompletion();
        } catch (MqttPersistenceException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (MqttException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        // Check that Message has been delivered
        info("Message Delivered: " + pubToken.isComplete());
        info("=== send end.====");
    }

    void disconnect() {
        IMqttToken disconnectToken;
        try {
            disconnectToken = client.disconnect();
            disconnectToken.waitForCompletion();
            client.close();
        } catch (MqttException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        client = null;
    }

    void info(String s) {
        System.out.println(s);
    }

    public void connectionLost(Throwable thrwbl) {
        // TODO Auto-generated method stub
        info("connectionLost");

        info("MQTT is disconnected from topic: {}. Message: {}. Cause: {}" + topic + thrwbl.getMessage() + thrwbl.getCause().getMessage());
        thrwbl.printStackTrace();

    }

    public void deliveryComplete(IMqttDeliveryToken arg0) {
        // TODO Auto-generated method stub
        info("deliveryComplete");

    }

    public void messageArrived(String arg0, MqttMessage arg1) throws Exception {
        // TODO Auto-generated method stub
        String message = new String(arg1.getPayload());
        String topic = arg0;

        info("xxx Receive : topic=" + topic + "; message=" + message);

    }
}

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.italktv.mqtt.client</groupId>
    <artifactId>mqttclient</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>mqttclient</name>
    <url>http://maven.apache.org</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>3.8.1</version>
            <scope>test</scope>
        </dependency>


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

    </dependencies>
</project>

上面是maven管理項目的pom.xml