参看官方文档:
* Java 为 MQ Telemetry Transport 创建异步发布程序
*在此任务中,您将遵循教程来修改第一个发布程序。通过修改,
*使应用程序能够发送发布而不等待传递确认信息。传递确认
*信息由您创建的回调类来接收。
*
*
*
*4.使客户机断开连接
* a.除去其中包含 token.waitForCompletion 表达式的语句。 主线程将继续执行,而不等待传递发布。
* b.测试客户机是否已断开连接。 将错误返回到 MqttCallback 中的 lostConnection 方法之后,MQTT 客户机将断开连接,客户机应用程序也可能断开连接。测试是否有打开的连接。
* c.使用常量 Example.quiesceTimeout 来设置使客户机停顿的最长时间。
* if (client.isConnected())
* client.disconnect(Example.quiesceTimeout);
*当满足下面三种情况的组合形式时,客户机就完成了:
* a.已经对在此会话中(如果重新启动了会话,则是在先前会话中)已发布的所有消息调用了回调。
* b.消息未完成,然而停顿时间间隔已到期。缺省情况下,停顿时间间隔为 30 秒。通过将要等待的毫秒数作为 client.disconnect 的一个参数来传递,即可更改停顿超时。
* c.在发布了某些消息并由客户机进行排队之后,但是在发送这些消息之前调用了 client.disconnect。已排队的消息尚未处于“未完成”状态。如果会话可重新启动,那么重新启动会话时就会重新发送消息。
* 缺省情况下,停顿时间间隔为 30 秒。
MQTT的消息发布代码:
package com.etrip.wsmqtt.server; import com.ibm.micro.client.mqttv3.MqttClient; import com.ibm.micro.client.mqttv3.MqttDeliveryToken; import com.ibm.micro.client.mqttv3.MqttMessage; import com.ibm.micro.client.mqttv3.MqttTopic; /** * 使用 Java 为 MQ Telemetry Transport 创建异步发布程序 * * * * * 消息发布的类的具体的实现 * * @author longgangbai * */ public class WSMQTTServerPubAsync { public static void main(String[] args) { try { //创建MqttClient对象 MqttClient client = new MqttClient(WSMQTTServerCommon.TCPAddress, WSMQTTServerCommon.clientId); //创建MQTT相关的主题 MqttTopic topic = client.getTopic(WSMQTTServerCommon.topicString); //创建MQTT的消息体 MqttMessage message = new MqttMessage(); //设置消息传输的类型 message.setQos(2); //设置是否在服务器中保存消息体 message.setRetained(false); //设置消息的内容 message.setPayload(WSMQTTServerCommon.publication.getBytes()); //创建一个MQTT的回调类 WSMQTTServerCallBack callback = new WSMQTTServerCallBack(WSMQTTServerCommon.clientId); //MqttClient绑定 client.setCallback(callback); //MqttClient连接 client.connect(); System.out.println("Publishing \"" + message.toString() + "\" on topic \"" + topic.getName() + "\" with QoS = " + message.getQos()); System.out.println("For client instance \"" + client.getClientId() + "\" on address " + client.getServerURI() + "\""); //发送消息并获取回执 MqttDeliveryToken token = topic.publish(message); System.out.println("With delivery token \"" + token.hashCode() + " delivered: " + token.isComplete()); Thread.sleep(100000000000000l); //关闭连接 if (client.isConnected()) client.disconnect(WSMQTTServerCommon.quiesceTimeout); System.out.println("Disconnected: delivery token \"" + token.hashCode() + "\" received: " + token.isComplete()); } catch (Exception e) { e.printStackTrace(); } } }
MQTT消息发布回调代码:
package com.etrip.wsmqtt.server; import com.ibm.micro.client.mqttv3.*; /** * 发布消息的回调类 * * 必须实现MqttCallback的接口并实现对应的相关接口方法 * ◦CallBack 类将实现 MqttCallBack。每个客户机标识都需要一个回调实例。在此示例中,构造函数传递客户机标识以另存为实例数据。在回调中,将它用来标识已经启动了该回调的哪个实例。 * ◦必须在回调类中实现三个方法: * * public void messageArrived(MqttTopic topic, MqttMessage message) * 接收已经预订的发布。 * * public void connectionLost(Throwable cause) * 在断开连接时调用。 * * public void deliveryComplete(MqttDeliveryToken token)) * 接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。 * * * ◦由 MqttClient.connect 激活此回调。 * * @author longgangbai */ public class WSMQTTServerCallBack implements MqttCallback { private String instanceData = ""; public WSMQTTServerCallBack(String instance) { instanceData = instance; } /** * 接收到消息的回调的方法 */ public void messageArrived(MqttTopic topic, MqttMessage message) { try { System.out.println("Message arrived: \"" + message.toString() + "\" on topic \"" + topic.toString() + "\" for instance \"" + instanceData + "\""); } catch (Exception e) { e.printStackTrace(); } } /** * 消息连接丢失 */ public void connectionLost(Throwable cause) { System.out.println("Connection lost on instance \"" + instanceData + "\" with cause \"" + cause.getMessage() + "\" Reason code " + ((MqttException)cause).getReasonCode() + "\" Cause \"" + ((MqttException)cause).getCause() + "\""); cause.printStackTrace(); } /** * */ public void deliveryComplete(MqttDeliveryToken token) { try { System.out.println("Delivery token \"" + token.hashCode() + "\" received by instance \"" + instanceData + "\""); } catch (Exception e) { e.printStackTrace(); } } }
常量类:
package com.etrip.wsmqtt.server; import java.util.UUID; /** * * 消息发布消息的常量字段 * * @author longgangbai */ public final class WSMQTTServerCommon { //发布broker的ip和端口 public static final String TCPAddress =System.getProperty("TCPAddress", "tcp://192.168.208.46:1883"); //客户端的Id public static String clientId =String.format("%-23.23s", System.getProperty("clientId", (UUID.randomUUID().toString())).trim()).replace('-', '_'); //发布消息的主题 public static final String topicString = System.getProperty("topicString", "china/beijing"); //发布的消息 public static final String publication =System.getProperty("publication", "Hello World " + String.format("%tc", System.currentTimeMillis())); //超时时间 public static final int quiesceTimeout = Integer.parseInt(System.getProperty("timeout", "10000")); public static final int sleepTimeout = Integer.parseInt(System.getProperty("timeout", "10000")); public static final boolean cleanSession =Boolean.parseBoolean(System.getProperty("cleanSession", "false")); public static final int QoS =Integer.parseInt(System.getProperty("QoS", "1")); public static final boolean retained =Boolean.parseBoolean(System.getProperty("retained", "false")); }
相关推荐
支持 MQTT server 服务端。 支持 MQTT 遗嘱消息。 支持 MQTT 保留消息。 支持自定义消息(mq)处理转发实现集群。 MQTT 客户端 阿里云 mqtt 连接 demo。 支持 GraalVM 编译成本机可执行程序。 支持 Spring ...
MQTTnet在.net core(2.2版本)开发环境下的实例代码,包含服务端,两个客户端(订阅、发布),并在阿里云的mqtt服务测试过可用
MQTT服务端,在本地直接打开可用,可配合之前发的MQTT客户端使用。
网上找的MQTT服务端的实现,代码有基本注释,自己研究一下
mqtt+服务端+客户端
mqtt服务端搭建到android使用教程 详细
Winform中使用MQTTnet实现MQTT的服务端和客户端之间的通信以及将订阅的消息保存到文件示例代码.rar Winform中使用MQTTnet实现MQTT的服务端和客户端之间的通信以及将订阅的消息保存到文件示例代码.rar Winform中使用...
MQTT协议的PHP服务端demo
C# 基于MQTTNet的服务端与客户端通信案例
目录 一、简介 ...MQTT 全称为 Message Queuing Telemetry Transport(消息队列遥测传输)是一种基于发布/订阅范式的“轻量级”消息协议,由 IBM 发布。 MQTT 可以被解释为一种低开销,低带宽占用
mqtt实现demo
官方下载的MQTTNet服务端实例用的基于MQTTNet的服务端,不支持Websocket交互,通过反复查阅资料,使用基于.Net Core(3.1) 和MQTTnet.AspNetCore(3.0.9)的服务端程序可以实现同时支持cs客户端、Websocket客户端的...
使用高性能的MQTTnet搭建的server和client例程,亲测可用。MQTTnet 是一个基于 MQTT 通信的高性能 .NET 开源库,它同时支持 MQTT 服务器端和客户端。
winForm中使用MQTT收发消息,不懂的可以联系qq:502701291
java中使用消息中间件ActiveMQ的MQTT协议发布消息使用fusesource,fusesource提供三种方式实现发布消息的方式,分别是阻塞式(BlockingConnection)、回调式(CallbackConnection)和Future样式(FutureConnection)
此Demo对应本人博客文章《MQTT(一)C#使用 MQTTnet 快速实现 MQTT 通信》 开发环境Win7 + vs2017
moquette对mqtt的实现,包括broke代理服务器的实现等。
Android开发,通过使用mqtt3.1.1版本和mqtt5.0版本进行mqtt数据的发送和接收,实现和服务端数据的交互
包含翻译后的API文档:mqtt-client-1.3-javadoc-API文档-中文(简体)版.zip 对应Maven信息:groupId:org.fusesource.mqtt-client,artifactId:mqtt-client,version:1.3 使用方法:解压翻译后的API文档,用浏览器...
MQTT(MessageQueuingTelemetryTransport,消息队列遥测传输)是IBM开发的一个即时通讯协议,它是一种轻量级的、基于代理的“发布/订阅”模式的消息传输协议。其具有协议简洁、小巧、可扩展性强、省流量、省电等优点...