FuseSource MQTT的实现发送和接收消(java 实现)

1.MQTT是什么？ MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的“轻量级”通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

2.代码实现 先引入jar包

<dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>1.2.0</version> </dependency> <!-- 引入另一种mqtt的client --> <dependency> <groupId>org.fusesource.mqtt-client</groupId> <artifactId>mqtt-client</artifactId> <version>1.14</version> </dependency>

本例子使用第二个依赖实现的

/** * @Auther: 18030501 * @Date: 2018/10/25 17:15 * @Description: 发布端/接收端 */ @Slf4j public class PublishAndSubscribeClient { // mqtt服务 public static String HOST = "tcp://host:port"; // 客户端id public static String CLIENT_ID = "test_clientId"; // 用户名 public static String USERNAME = "whale"; // 密码 public static String PASSWORD = "123456"; // 客户端与服务端消息传递的最大间隔。它能够使服务器检测到客户端的网络是否已经丢失，而无需等待TCP/IP超时 public static short KEEP_ALIVE = 20; public static void main(String[] args) throws Exception { // 配置MQTT连接 MQTT mqtt = new MQTT(); mqtt.setHost(HOST); mqtt.setClientId(CLIENT_ID); mqtt.setUserName(USERNAME); mqtt.setPassword(PASSWORD); // 是否清除session，false:MQTT服务器保存于客户端会话的的主题与确认位置,true:MQTT服务器不保存于客户端会话的的主题与确认位置 mqtt.setCleanSession(true); mqtt.setKeepAlive(KEEP_ALIVE); // 设置重新连接的次数 mqtt.setConnectAttemptsMax(1); // 设置重新连接的间隔 mqtt.setReconnectDelay(4); // 获取mqtt的连接对象BlockingConnection CallbackConnection connection = mqtt.callbackConnection(); // 添加接收消息的监听 connection.listener(new Listener() { @Override public void onConnected() { System.out.println("----onConnected----"); } @Override public void onDisconnected() { System.out.println("----onDisconnected----"); } @Override public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) { // 处理来自topic的消息 log.info("receive topic:{} and message:{}", topic.toString(), new String(payload.getData())); ack.run(); } @Override public void onFailure(Throwable throwable) { System.out.println("----onFailure----"); } }); // 添加连接的事件，并在连接成功时候订阅主题，发送消息 connection.connect(new Callback<Void>() { // 连接成功时执行的方法 @Override public void onSuccess(Void aVoid) { final String topic = "mqtt/test"; // 订阅主题 connection.subscribe(new Topic[]{new Topic(topic, QoS.AT_LEAST_ONCE)}, null); final String message = "hello everyone!"; log.info("MQTT CallbackServer publish topic={},message:{}", topic, message); // 参数说明：1.主题 2.消息 3.服务质量(0,1,2) 4.发布保留标识，表示服务器是否要保留发布的消息 5.发布成功回调 // 服务质量： // 0：最多一次，即：<=1 ; 1:至少一次，即：>=1; 2:一次，即：=1 // 发布消息 connection.publish(topic, message.getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() { public void onSuccess(Void v) { // the publish operation completed successfully. log.info("----publish message is success!----"); } public void onFailure(Throwable value) { log.info("----publish message is failure!----"); value.printStackTrace(); } }); try { Thread.sleep(20000); } catch (InterruptedException e) { e.printStackTrace(); } } // 连接失败时执行的方法 @Override public void onFailure(Throwable value) { // If we could not connect to the server. log.info("MQTTCallbackServer.CallbackConnection.connect.onFailure 连接失败......{}", value.getMessage()); value.printStackTrace(); } }); // 等待消息发送和处理 try { Thread.sleep(20000); } catch (InterruptedException e) { e.printStackTrace(); } // 断开连接 connection.disconnect(new Callback<Void>() { @Override public void onSuccess(Void aVoid) { log.info("--------disconnect success-----------"); } @Override public void onFailure(Throwable throwable) { log.info("--------disconnect failure-----------"); } }); } }

下面在提供一个使用第一个jar包实现发送和接收消息：

/** * @Auther: 18030501 * @Date: 2018/10/25 15:18 * @Description: 发布端 */ public class PublishExample { public static String topic = "mqtt/test"; public static String content = "我在测试mqtt消息发送"; public static String broker = "tcp://host:port"; public static String userName = "test"; public static String password = "test"; public static String clientId = "pubClient"; public static int qos = 1; // 内存存储 public static MemoryPersistence persistence = new MemoryPersistence(); public static void main(String[] args) { try { // 1.创建客户端 MqttClient client = new MqttClient(broker, clientId, persistence); // 2.创建连接参数 MqttConnectOptions options = new MqttConnectOptions(); // 3.在重新启动和重新连接时记住状态 options.setCleanSession(true); // 4.设置连接的用户名和密码 options.setUserName(userName); options.setPassword(password.toCharArray()); // 5.建立连接 client.connect(options); // 6.创建消息 MqttMessage message = new MqttMessage(content.getBytes()); // 7.设置消息的服务质量 message.setQos(qos); // 8.发布消息 client.publish(topic, message); // 9.断开连接 client.disconnect(); // 10.关闭客户端 client.close(); } catch (MqttException me) { System.out.println("reason " + me.getReasonCode()); System.out.println("msg " + me.getMessage()); System.out.println("loc " + me.getLocalizedMessage()); System.out.println("cause " + me.getCause()); System.out.println("excep " + me); me.printStackTrace(); } } } /** * @Auther: 18030501 * @Date: 2018/10/25 15:49 * @Description: 订阅端 */ public class SubscribeExample { public static String HOST = "tcp://10.245.39.91:61613"; public static String TOPIC = "mqtt/test"; public static int qos = 1; public static String clientId = "subClient"; public static String userName = "test"; public static String password = "test"; public static void main(String[] args) throws Exception { try { // 1.创建客户端 MqttClient client = new MqttClient(HOST, clientId, new MemoryPersistence()); // 2.Mqtt的连接设置 MqttConnectOptions options = new MqttConnectOptions(); options.setCleanSession(true); options.setUserName(userName); options.setPassword(password.toCharArray()); options.setConnectionTimeout(10); options.setKeepAliveInterval(20); // 3.设置回调函数 client.setCallback(new MqttCallback() { @Override public void connectionLost(Throwable throwable) { System.out.println("connectionLost"); } @Override public void messageArrived(String topic, MqttMessage message) throws Exception { System.out.println("topic:" + topic); System.out.println("Qos:" + message.getQos()); System.out.println("message content:" + new String(message.getPayload())); } @Override public void deliveryComplete(IMqttDeliveryToken token) { System.out.println("deliveryComplete---------" + token.isComplete()); } }); // 4.建立连接 client.connect(options); // 5.订阅消息 client.subscribe(TOPIC, qos); } catch (Exception e) { e.printStackTrace(); } } }