`
can_do
  • 浏览: 249789 次
  • 性别: Icon_minigender_1
  • 来自: 北京
社区版块
存档分类
最新评论

EMQ X(MQTT Broker)中如何使用持久化订阅

阅读更多
1、EMQ X支持持久化订阅
首先理解下,什么是持久化订阅?持久化订阅指的是,消费端无论是否在线,只要其持久化订阅了某个Topic,消费端就总会收到发送到此Topic上的消息,类似ActiveMQ中的持久化订阅。

2、持久化订阅使用方式,从以下客户端举例:

1> 以mosquitto客户端为例,
通过参数[-c]起作用,注意[-c]需要和[-i]结合使用

>mosquitto_sub -h 172.17.6.147 -t YourComTm/01012345678  -c -i "can_do@2019" -q 2 -u can_do -P passw0rd  -d
Client can_do@2019 sending CONNECT
Client can_do@2019 received CONNACK (0)
Client can_do@2019 sending SUBSCRIBE (Mid: 1, Topic: YourComTm/01012345678, QoS: 2, Options: 0x00)
Client can_do@2019 received PUBLISH (d0, q2, r0, m4, 'YourComTm/01012345678', ... (16 bytes))
Client can_do@2019 sending PUBREC (m4, rc0)
Client can_do@2019 received SUBACK
Subscribed (mid: 1): 2
Client can_do@2019 received PUBLISH (d0, q2, r1, m5, 'YourComTm/01012345678', ... (16 bytes))
Client can_do@2019 sending PUBREC (m5, rc0)
Client can_do@2019 received PUBREL (Mid: 4)
Client can_do@2019 sending PUBCOMP (m4)
hello can_do 994
Client can_do@2019 received PUBREL (Mid: 5)
Client can_do@2019 sending PUBCOMP (m5)
hello can_do 994

2> Java客户端mqtt-client中实现持久化订阅,

关键代码行:
mqtt.setClientId(this.clientId);
mqtt.setCleanSession(false);


public TestMqttClientSubMsgInFuture(String paramBrokerHost, int paramBrokerPort, String paramUserName,
			String paramPassword, String paramTopicName, String paramClientId) {
		this.brokerHost = paramBrokerHost;
		this.brokerPort = paramBrokerPort;
		this.username = paramUserName;
		this.password = paramPassword;
		this.topicName = paramTopicName;
		this.clientId = paramClientId;

		MQTT mqtt = new MQTT();
		try {
			// set clientId and cleanSessionFlag in mqtt object
			mqtt.setHost(this.brokerHost, this.brokerPort);
			mqtt.setUserName(this.username);
			mqtt.setPassword(this.password);
			mqtt.setClientId(this.clientId);
			mqtt.setCleanSession(false);
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		connection = mqtt.futureConnection();

		try {
			connection.connect();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		try {
			connection.subscribe(new Topic[] { new Topic(this.topicName, QoS.EXACTLY_ONCE) });

		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

3、注意事项:
1> 在EMQX中,session是通过clientId标识的,因此clientId在集群全局中要保持唯一性,否则,可能消费不了消息,或者消费消息不是预期效果。

【温馨提示】
如果您觉得满意,可以选择支持下,您的支持是我最大的动力:

分享到:
评论

相关推荐

Global site tag (gtag.js) - Google Analytics