代码：
package cc.gongchang.mqtt; import java.net.URISyntaxException; import org.fusesource.hawtdispatch.Dispatch; import org.fusesource.mqtt.client.Future; import org.fusesource.mqtt.client.FutureConnection; import org.fusesource.mqtt.client.MQTT; import org.fusesource.mqtt.client.Message; import org.fusesource.mqtt.client.QoS; import org.fusesource.mqtt.client.Topic; import com.alibaba.fastjson.JSONObject; /** * Hello world! * */ public class App { public static void main(String[] args) { MQTT mqtt = new MQTT(); // MQTT设置说明 // 设置主机号 try { mqtt.setHost("tcp://sgdzpic.3322.org:1883"); } catch (URISyntaxException e) { // TODO Auto-generated catch block e.printStackTrace(); } // 用于设置客户端会话的ID。在setCleanSession(false);被调用时，MQTT服务器利用该ID获得相应的会话。此ID应少于23个字符，默认根据本机地址、端口和时间自动生成 mqtt.setClientId("876543210"); // 若设为false，MQTT服务器将持久化客户端会话的主体订阅和ACK位置，默认为true mqtt.setCleanSession(false); // 定义客户端传来消息的最大时间间隔秒数，服务器可以据此判断与客户端的连接是否已经断开，从而避免TCP/IP超时的长时间等待 mqtt.setKeepAlive((short) 60); // 服务器认证用户名 mqtt.setUserName("admin"); // 服务器认证密码 mqtt.setPassword("123456"); // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息 mqtt.setWillTopic("willTopic"); // 设置“遗嘱”消息的内容，默认是长度为零的消息 mqtt.setWillMessage("willMessage"); // 设置“遗嘱”消息的QoS，默认为QoS.ATMOSTONCE mqtt.setWillQos(QoS.AT_LEAST_ONCE); // 若想要在发布“遗嘱”消息时拥有retain选项，则为true mqtt.setWillRetain(true); // 设置版本 mqtt.setVersion("3.1.1"); // 失败重连接设置说明 // 客户端首次连接到服务器时，连接的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1 mqtt.setConnectAttemptsMax(10L); // 客户端已经连接到服务器，但因某种原因连接断开时的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1 mqtt.setReconnectAttemptsMax(3L); // 首次重连接间隔毫秒数，默认为10ms mqtt.setReconnectDelay(10L); // 重连接间隔毫秒数，默认为30000ms mqtt.setReconnectDelayMax(30000L); // 设置重连接指数回归。设置为1则停用指数回归，默认为2 mqtt.setReconnectBackOffMultiplier(2); // Socket设置说明 // 设置socket接收缓冲区大小，默认为65536（64k） mqtt.setReceiveBufferSize(65536); // 设置socket发送缓冲区大小，默认为65536（64k） mqtt.setSendBufferSize(65536); // 设置发送数据包头的流量类型或服务类型字段，默认为8，意为吞吐量最大化传输 mqtt.setTrafficClass(8); // 带宽限制设置说明 // 设置连接的最大接收速率，单位为bytes/s。默认为0，即无限制 mqtt.setMaxReadRate(0); // 设置连接的最大发送速率，单位为bytes/s。默认为0，即无限制 mqtt.setMaxWriteRate(0); Boolean contition = true; while (contition) { // 选择消息分发队列 // 若没有调用方法setDispatchQueue，客户端将为连接新建一个队列。如果想实现多个连接使用公用的队列，显式地指定队列是一个非常方便的实现方法 mqtt.setDispatchQueue(Dispatch.createQueue("foo")); FutureConnection connection = mqtt.futureConnection(); Future<Void> f1 = connection.connect(); try { f1.await(); } catch (Exception e1) { // TODO Auto-generated catch block e1.printStackTrace(); } // 订阅消息 Future<byte[]> f2 = connection.subscribe(new Topic[] { new Topic("person/blacklist/#", QoS.AT_LEAST_ONCE) }); // try { byte[] qoses = f2.await(); } catch (Exception e1) { // TODO Auto-generated catch block e1.printStackTrace(); } // 发送身份验证消息. // Future<Void> f3 = connection.publish("foo", "Hello".getBytes(), // QoS.AT_LEAST_ONCE, false); // 接收订阅消息.. Future<Message> receive = connection.receive(); // 打印消息. Message message = null; try { message = receive.await(); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } String subscribeInfo = String.valueOf(message.getPayloadBuffer()); if(subscribeInfo.startsWith("ascii")) { JSONObject subscribeObject = JSONObject.parseObject(subscribeInfo.substring(7)); System.out.println(subscribeInfo); System.out.println(subscribeObject.get("name"));//姓名，对应name System.out.println(subscribeObject.get("certifiedNo"));//身份证，对应identityId System.out.println(subscribeObject.get("url"));//图片地址，对应imageUrl //还需要传送tarLibSerial //入库黑名单 } // 回应 message.ack(); // Future<Void> f4 = connection.disconnect(); try { f4.await(); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } } } }