MQTT服务器的搭建与测试pub/sub通信过程(2)

Maven依赖为:

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

订阅(Sub)端HelloWorld:

订阅一个名为foo的主题，消息级别为 AT_LEAST_ONCE

MQTT mqtt = new MQTT(); try { mqtt.setHost("tcp://192.168.2.112:1883"); System.out.println("start"); BlockingConnection connection = mqtt.blockingConnection(); connection.connect(); System.out.println(connection == null); connection.subscribe(new Topic[] { new Topic("foo", QoS.AT_LEAST_ONCE) }); while (true) { Message message = connection.receive(); System.out.println("MQTTFutureClient.Receive Message " + "Topic Title :" + message.getTopic() + " context :" + String.valueOf(message.getPayloadBuffer())); } } catch (Exception e) { } System.out.println("end");

发布(pub)端：

给一个名为foo的主题推送消息，消息级别为 AT_LEAST_ONCE

MQTT mqtt = new MQTT(); try { mqtt.setHost("tcp://192.168.2.112:1883"); System.out.println("start"); BlockingConnection connection = mqtt.blockingConnection(); connection.connect(); System.out.println(connection == null); for (int i = 0; i <100000; i++) { connection.publish("foo", "HelloWQEQWEQ".getBytes(), QoS.AT_LEAST_ONCE, false); } } catch (Exception e) { } System.out.println("end");

通过运行便可以看到消息成功pub到sub端。

至于连接类型，该客户端提供了三种

BlockingConnection 阻塞式

CallbackConnection 回调函数式

FutureConnection 异步式