MQTT客户端代码 -- JAVA fusesource mqtt-client

2023年1月30日 265次阅读 来源: 绝尘的神马

转：http://ywx217.iteye.com/blog/1797463

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();
connection.publish("foo", "Hello".toBytes(), QoS.AT_LEAST_ONCE, false);

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

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

　　

    原文作者：绝尘的神马
    原文地址: https://www.cnblogs.com/hexinlin/articles/6507348.html
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞