MQTT客户端代码

 

MQTT是一款针对机对机（M2M）通信的，非常轻量级的的消息订阅、发布协议。它适用于一些系统资源和网络带宽非常有限的情况下的远程连接。MQTT-Client提供一个ASL 2.0证书下的MQTT接口。在网络连接失败时，它能够自动地重新连接服务器并尝试恢复会话。应用程序能够使用阻塞API、基于Future的API和回调API，共三种接口形式。

 

在Maven中引用MQTT-Client

将下列文本加入到pom.xml文件中。

<dependency>
  <groupId>org.fusesource.mqtt-client</groupId>
  <artifactId>mqtt-client</artifactId>
  <version>1.0-SNAPSHOT</version>
</dependency>

<repositories>
  <repository>
    <id>fusesource.snapshots</id>
    <name>FuseSource Snapshot Repository</name>
    <url>http://repo.fusesource.com/nexus/content/repositories/snapshots</url>
    <snapshots><enabled>true</enabled></snapshots>
    <releases><enabled>false</enabled></releases>
  </repository>
</repositories>

 

以其他方式引用MQTT-Client

下载uber jar文件并加入编译路径中，该jar文件中包含MQTT-Client的所有依赖。

在Java 1.4环境中使用

作者同时提供了适用于Java 1.4的API。由于Java1.4中没有SSLEngine类依赖的NIO，因此该API不支持SSL连接。

 

配置MQTT连接

前面提到的阻塞、Future和回调这3种API，建立连接时使用的代码时完全相同的。首先新建一个MQTT对象并配置连接参数。在连接前至少要调用setHost方法，来指定所要连接的服务器地址。

MQTT mqtt = new MQTT(); 
mqtt.setHost("localhost", 1883); 
// or 
mqtt.setHost("tcp://localhost:1883");

 

MQTT设置说明

• setClientId：用于设置客户端会话的ID。在setCleanSession(false);被调用时，MQTT服务器利用该ID获得相应的会话。此ID应少于23个字符，默认根据本机地址、端口和时间自动生成。
• setCleanSession：若设为false，MQTT服务器将持久化客户端会话的主体订阅和ACK位置，默认为true。
• setKeepAlive：定义客户端传来消息的最大时间间隔秒数，服务器可以据此判断与客户端的连接是否已经断开，从而避免TCP/IP超时的长时间等待。
• setUserName：服务器认证用户名。
• setPassword：服务器认证密码。
• setWillTopic：设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
• setWillMessage：设置“遗嘱”消息的内容，默认是长度为零的消息。
• setWillQos：设置“遗嘱”消息的QoS，默认为QoS.ATMOSTONCE。
• setWillRetain：若想要在发布“遗嘱”消息时拥有retain选项，则为true。

失败重连接设置说明

网络出现故障时，程序能够自动重新连接并重建会话。利用下列方法能够配置重新连接的间隔和最大重试次数：

• setConnectAttemptsMax：客户端首次连接到服务器时，连接的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1。
• setReconnectAttemptsMax：客户端已经连接到服务器，但因某种原因连接断开时的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1。
• setReconnectDelay：首次重连接间隔毫秒数，默认为10ms。
• setReconnectDelayMax：重连接间隔毫秒数，默认为30000ms。
• setReconnectBackOffMultiplier：设置重连接指数回归。设置为1则停用指数回归，默认为2。

Socket设置说明

可以利用下列方法调整socket设置：

• setReceiveBufferSize：设置socket接收缓冲区大小，默认为65536（64k）。

• setSendBufferSize：设置socket发送缓冲区大小，默认为65536（64k）。

• setTrafficClass：设置发送数据包头的流量类型或服务类型字段，默认为8，意为吞吐量最大化传输。

带宽限制设置说明

可通过下述方法设置读写速率限制：

• setMaxReadRate：设置连接的最大接收速率，单位为bytes/s。默认为0，即无限制。

• setMaxWriteRate：设置连接的最大发送速率，单位为bytes/s。默认为0，即无限制。

使用SSL连接

如果想使用SSL/TLS连接，替代TCP连接，可以使用“ssl://”或“tls://”作为连接URI前缀，实现安全连接。支持的协议包括：

• ssl:// - 使用JVM默认版本的SSL算法。
• sslv*:// - 使用指定版本的SSL算法，星号为JVM支持的SSL算法版本，例如sslv3。
• tls:// - 使用JVM默认版本的TLS算法。
• tlsv*:// - 使用指定版本的TLS算法，星号为JVM支持的TLS算法版本，例如tlsv1.1。

客户端使用JVM的SSLContext，基于在JVM的系统配置进行连接。可以调用MQTT的setSslContext方法，换用其他连接方式。
对于内部线程池，SSL连接为阻塞特性。调用setBlockingExecutor方法可以替换所要使用的executor。

选择消息分发队列

若没有调用方法setDispatchQueue，客户端将为连接新建一个队列。如果想实现多个连接使用公用的队列，显式地指定队列是一个非常方便的实现方法。
 

使用阻塞API

MQTT.connectBlocking方法建立并返回一个阻塞API连接。

BlockingConnection connection = mqtt.blockingConnection();
connection.connect();

使用publish方法发布消息：

connection.publish("foo", "Hello".toBytes(), QoS.AT_LEAST_ONCE, false);

可以利用subscribe方法订阅多个主题：

Topic[] topics = {new Topic("foo", QoS.AT_LEAST_ONCE)};
byte[] qoses = connection.subscribe(topics);

利用receive和ack方法，获取并应答消息：

Message message = connection.receive();
System.out.println(message.getTopic());
byte[] payload = message.getPayload();
// process the message then:
message.ack();

结束连接：

connection.disconnect();

 

使用基于Future的API

MQTT.connectFuture方法建立并返回一个基于Future类型的API连接。所有连接操作都是非阻塞的，连接结果通过Future对象返回。

FutureConnection connection = mqtt.futureConnection();
Future<Void> f1 = connection.connect();
f1.await();

Future<byte[]> f2 = connection.subscribe(new Topic[]{new Topic(utf8("foo"), QoS.AT_LEAST_ONCE)});
byte[] qoses = f2.await();

// We can start future receive..
Future<Message> receive = connection.receive();

// send the message..
Future<Void> f3 = connection.publish("foo", "Hello".getBytes(), QoS.AT_LEAST_ONCE, false);

// Then the receive will get the message.
Message message = receive.await();
message.ack();

Future<Void> f4 connection.disconnect();
f4.await();

 

使用回调API

MQTT.connectCallback方法建立并返回一个回调API连接。本方法是三种API中最复杂也是性能最好的API，前面提到的两种API其实都是对回调API的封装。对连接的所有操作都是非阻塞的，返回的结果将传至回调接口函数。示例如下：

final CallbackConnection connection = mqtt.callbackConnection();
connection.listener(new Listener() {

    public void onDisconnected() {
    }
    public void onConnected() {
    }

    public void onSuccess(UTF8Buffer topic, Buffer payload, Runnable ack) {
        // You can now process a received message from a topic.
        // Once process execute the ack runnable.
        ack.run();
    }
    public void onFailure(Throwable value) {
        connection.close(null); // a connection failure occured.
    }
})
connection.connect(new Callback<Void>() {
    public void onFailure(Throwable value) {
        result.failure(value); // If we could not connect to the server.
    }

    // Once we connect..
    public void onSuccess(Void v) {

        // Subscribe to a topic
        Topic[] topics = {new Topic("foo", QoS.AT_LEAST_ONCE)};
        connection.subscribe(topics, new Callback<byte[]>() {
            public void onSuccess(byte[] qoses) {
                // The result of the subcribe request.
            }
            public void onFailure(Throwable value) {
                connection.close(null); // subscribe failed.
            }
        });

        // Send a message to a topic
        connection.publish("foo", "Hello".getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
            public void onSuccess(Void v) {
              // the pubish operation completed successfully.
            }
            public void onFailure(Throwable value) {
                connection.close(null); // publish failed.
            }
        });

        // To disconnect..
        connection.disconnect(new Callback<Void>() {
            public void onSuccess(Void v) {
              // called once the connection is disconnected.
            }
            public void onFailure(Throwable value) {
              // Disconnects never fail.
            }
        });
    }
});

Every connection has a HawtDispatch dispatch queue which it uses to process IO events for the socket. The dispatch queue is an Executor that provides serial execution of IO and processing events and is used to ensure synchronized access of connection.

The callbacks will be executing the the the dispatch queue associated with the connection so it safe to use the connection from the callback but you MUST NOT perform any blocking operations within the callback. If you need to perform some processing which MAY block, you must send it to another thread pool for processing. Furthermore, if another thread needs to interact with the connection it can only doit by using a Runnable submitted to the connection's dispatch queue.

Example of executing a Runnable on the connection's dispatch queue:

connection.getDispatchQueue().execute(new Runnable(){
    public void run() {
      connection.publish( ..... );
    }
});

 

相关链接：

• 原文：http://mqtt-client.fusesource.org/
• MQTT 3.1文档：http://mqtt.org/documentation