代码：
1package cc.gongchang.mqtt; 2 3import java.net.URISyntaxException; 4 5import org.fusesource.hawtdispatch.Dispatch; 6import org.fusesource.mqtt.client.Future; 7import org.fusesource.mqtt.client.FutureConnection; 8import org.fusesource.mqtt.client.MQTT; 9import org.fusesource.mqtt.client.Message; 10import org.fusesource.mqtt.client.QoS; 11import org.fusesource.mqtt.client.Topic; 12 13import com.alibaba.fastjson.JSONObject; 14 15/** 16 * Hello world! 17 * 18 */ 19public class App { 20 public static void main(String[] args) { 21 MQTT mqtt = new MQTT(); 22 // MQTT设置说明 23 // 设置主机号 24 try { 25 mqtt.setHost("tcp://sgdzpic.3322.org:1883"); 26 } catch (URISyntaxException e) { 27 // TODO Auto-generated catch block 28 e.printStackTrace(); 29 } 30 // 用于设置客户端会话的ID。在setCleanSession(false);被调用时，MQTT服务器利用该ID获得相应的会话。此ID应少于23个字符，默认根据本机地址、端口和时间自动生成 31 mqtt.setClientId("876543210"); 32 // 若设为false，MQTT服务器将持久化客户端会话的主体订阅和ACK位置，默认为true 33 mqtt.setCleanSession(false); 34 // 定义客户端传来消息的最大时间间隔秒数，服务器可以据此判断与客户端的连接是否已经断开，从而避免TCP/IP超时的长时间等待 35 mqtt.setKeepAlive((short) 60); 36 // 服务器认证用户名 37 mqtt.setUserName("admin"); 38 // 服务器认证密码 39 mqtt.setPassword("123456"); 40 // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息 41 mqtt.setWillTopic("willTopic"); 42 // 设置“遗嘱”消息的内容，默认是长度为零的消息 43 mqtt.setWillMessage("willMessage"); 44 // 设置“遗嘱”消息的QoS，默认为QoS.ATMOSTONCE 45 mqtt.setWillQos(QoS.AT_LEAST_ONCE); 46 // 若想要在发布“遗嘱”消息时拥有retain选项，则为true 47 mqtt.setWillRetain(true); 48 // 设置版本 49 mqtt.setVersion("3.1.1"); 50 // 失败重连接设置说明 51 // 客户端首次连接到服务器时，连接的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1 52 mqtt.setConnectAttemptsMax(10L); 53 // 客户端已经连接到服务器，但因某种原因连接断开时的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1 54 mqtt.setReconnectAttemptsMax(3L); 55 // 首次重连接间隔毫秒数，默认为10ms 56 mqtt.setReconnectDelay(10L); 57 // 重连接间隔毫秒数，默认为30000ms 58 mqtt.setReconnectDelayMax(30000L); 59 // 设置重连接指数回归。设置为1则停用指数回归，默认为2 60 mqtt.setReconnectBackOffMultiplier(2); 61 62 // Socket设置说明 63 // 设置socket接收缓冲区大小，默认为65536（64k） 64 mqtt.setReceiveBufferSize(65536); 65 // 设置socket发送缓冲区大小，默认为65536（64k） 66 mqtt.setSendBufferSize(65536); 67 // 设置发送数据包头的流量类型或服务类型字段，默认为8，意为吞吐量最大化传输 68 mqtt.setTrafficClass(8); 69 70 // 带宽限制设置说明 71 // 设置连接的最大接收速率，单位为bytes/s。默认为0，即无限制 72 mqtt.setMaxReadRate(0); 73 // 设置连接的最大发送速率，单位为bytes/s。默认为0，即无限制 74 mqtt.setMaxWriteRate(0); 75 76 Boolean contition = true; 77 while (contition) { 78 // 选择消息分发队列 79 // 若没有调用方法setDispatchQueue，客户端将为连接新建一个队列。如果想实现多个连接使用公用的队列，显式地指定队列是一个非常方便的实现方法 80 mqtt.setDispatchQueue(Dispatch.createQueue("foo")); 81 FutureConnection connection = mqtt.futureConnection(); 82 Future<Void> f1 = connection.connect(); 83 try { 84 f1.await(); 85 } catch (Exception e1) { 86 // TODO Auto-generated catch block 87 e1.printStackTrace(); 88 } 89 // 订阅消息 90 Future<byte[]> f2 = connection.subscribe(new Topic[] { 91 new Topic("person/blacklist/#", QoS.AT_LEAST_ONCE) }); 92 // 93 try { 94 byte[] qoses = f2.await(); 95 } catch (Exception e1) { 96 // TODO Auto-generated catch block 97 e1.printStackTrace(); 98 } 99 100 // 发送身份验证消息. 101 // Future<Void> f3 = connection.publish("foo", "Hello".getBytes(), 102 // QoS.AT_LEAST_ONCE, false); 103 // 接收订阅消息.. 104 Future<Message> receive = connection.receive(); 105 // 打印消息. 106 Message message = null; 107 try { 108 message = receive.await(); 109 } catch (Exception e) { 110 // TODO Auto-generated catch block 111 e.printStackTrace(); 112 } 113 String subscribeInfo = String.valueOf(message.getPayloadBuffer()); 114 if(subscribeInfo.startsWith("ascii")) { 115 JSONObject subscribeObject = JSONObject.parseObject(subscribeInfo.substring(7)); 116 System.out.println(subscribeInfo); 117 System.out.println(subscribeObject.get("name"));//姓名，对应name 118 System.out.println(subscribeObject.get("certifiedNo"));//身份证，对应identityId 119 System.out.println(subscribeObject.get("url"));//图片地址，对应imageUrl 120 //还需要传送tarLibSerial 121 //入库黑名单 122 } 123 // 回应 124 message.ack(); 125 // 126 Future<Void> f4 = connection.disconnect(); 127 try { 128 f4.await(); 129 } catch (Exception e) { 130 // TODO Auto-generated catch block 131 e.printStackTrace(); 132 } 133 } 134 135 } 136}