使用paho的MQTT时遇到的重连导致订阅无法收到问题和解决

简介:   最近在使用MQTT来实现消息的传输,网上demo很多,这里就不在重复介绍了,直接上代码,百度就能出现一大堆  下面是MQTT实现订阅的主要代码部分  MqttClient client = new MqttClient(HOST, clientid, new MemoryPersiste...

  最近在使用MQTT来实现消息的传输,网上demo很多,这里就不在重复介绍了,直接上代码,百度就能出现一大堆

  下面是MQTT实现订阅的主要代码部分

  

MqttClient client = new MqttClient(HOST, clientid, new MemoryPersistence());

MqttConnectOptions options = new MqttConnectOptions();

options.setCleanSession(true);

options.setUserName(userName);

options.setPassword(passWord.toCharArray());


options.setConnectionTimeout(10);


options.setKeepAliveInterval(20);
//设置断开后重新连接		
options.setAutomaticReconnect(true);

try {

               client.setCallback(new PushCallback());//设置各种情况的回调函数

               client.connect(options);
         
               //订阅消息
               int[] Qos  = {0};

   		String[] topic1 = {TOPIC};

   		client.subscribe(topic1, Qos);

        } catch (Exception e) {

               e.printStackTrace();

        }    

回到方法实现代码如下

 

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;

public class PushCallback implements MqttCallback {
	public void connectionLost(Throwable cause) {

		// 连接丢失后,触发这个方法

	}



	@Override
	public void deliveryComplete(IMqttDeliveryToken token) {
		// publish后会执行到这里
		
		
	}
	@Override
	public void messageArrived(String arg0, MqttMessage arg1) throws Exception {
		System.out.println("接收消息主题:"+arg0);

		System.out.println("接收消息Qos:"+arg1.getQos());

		System.out.println("接收消息内容:"+new String(arg1.getPayload()));
		
	}
}

  如果MQTT连接断开,会调用connectionLost 函数,我一开始觉得直接在这里使用client.connect(options),就可以直接实现重连了,结果报错mqtt的状态为 

这个好解决,如果这样设置,可以实现断开自动重连

//设置断开后重新连接
options.setAutomaticReconnect(true);

但这样重连是实现了,但是之前订阅的主题却接收不到消息了,需要重新订阅主题才能正常接收消息,那我这个重新订阅的代码要怎么再放进去呢,反正不是再connectionLost里就是了,那是后连接还没有重连连上!

继续看MQTT的connec的源码发现了一段代码使我找到了解决方案

MqttReconnectCallback 是实现MqttCallbackExtended接口的

发现comms中有设置重连的回调对象,但是怎么把这个回调由我们来主动放进去呢?继续往下看源码可以发现

 

也就是如果我们在之前放入client的回调对象是实现的 MqttCallbackExtended 接口,则MQTT会将我们的回调对象放入 connectActionListener 中 然后由 connectActionListener实现具体的connect

接下来我们不callback 对象改为实现 MqttCallbackExtended这个接口,然后实现下面方法,

@Override
public void connectComplete(boolean reconnect, String serverURI) {
		//连接成功后调用
      client.subscribe(topics,Qos);//具体订阅代码
}

  

 就可以解决MQTT重连后无法订阅的问题

 

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
物联网 Java Linux
Linux安装与配置Eclipse Paho库:实现MQTT通信
Eclipse Paho是一个开源的MQTT(Message Queuing Telemetry Transport)实现,提供了多种编程语言的客户端库,包括C、C++、Java、Python等。在Linux系统中,通过安装和配置Eclipse Paho库,我们可以方便地实现MQTT通信功能。本文将详细介绍在Linux系统中安装和配置Eclipse Paho库的步骤,以便于开发者在物联网项目中使用MQTT协议进行通信。
1291 0
|
消息中间件 Java 物联网
一文搞懂MQTT,如何在SpringBoot中使用MQTT实现消息的订阅和发布
之前介绍了RabbitMQ以及如何在SpringBoot项目中整合使用RabbitMQ,看过的朋友都说写的比较详细,希望再总结一下目前比较流行的MQTT。所以接下来,就来介绍什么MQTT?它在IoT中有着怎样的作用?如何在项目中使用MQTT?
13464 5
一文搞懂MQTT,如何在SpringBoot中使用MQTT实现消息的订阅和发布
|
3天前
|
消息中间件 测试技术
通过轻量消息队列(原MNS)主题HTTP订阅+ARMS实现自定义数据多渠道告警
轻量消息队列(原MNS)以其简单队列模型、轻量化协议及按量后付费模式,成为阿里云产品间消息传输首选。本文通过创建主题、订阅、配置告警集成等步骤,展示了该产品在实际应用中的部分功能,确保消息的可靠传输。
14 2
|
监控 物联网 API
【.NET+MQTT】.NET6 环境下实现MQTT通信,以及服务端、客户端的双边消息订阅与发布的代码演示
MQTT广泛应用于工业物联网、智能家居、各类智能制造或各类自动化场景等。MQTT是一个基于客户端-服务器的消息发布/订阅传输协议,在很多受限的环境下,比如说机器与机器通信、机器与物联网通信等。好了,科普的废话不多说,下面直接通过.NET环境来实现一套MQTT通信demo,实现服务端与客户端的双边消息发布与订阅的功能和演示。
1274 0
【.NET+MQTT】.NET6 环境下实现MQTT通信,以及服务端、客户端的双边消息订阅与发布的代码演示
|
5月前
|
消息中间件 存储 开发工具
消息队列 MQ产品使用合集之C++如何使用Paho MQTT库进行连接、发布和订阅消息
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
消息中间件 存储 负载均衡
两个实验让我彻底弄懂了「订阅关系一致」
这篇文章,笔者想聊聊 RocketMQ 最佳实践之一:**保证订阅关系一致**。 订阅关系一致指的是同一个消费者 Group ID 下所有 Consumer 实例所订阅的 Topic 、Tag 必须完全一致。 如果订阅关系不一致,消息消费的逻辑就会混乱,甚至导致消息丢失。
两个实验让我彻底弄懂了「订阅关系一致」
|
传感器 负载均衡 物联网
MQTT v5共享订阅是怎么回事?如何使用共享订阅提高消息订阅的灵活性和可伸缩性?
MQTT v5共享订阅是怎么回事?如何使用共享订阅提高消息订阅的灵活性和可伸缩性?
477 1
|
6月前
|
消息中间件 Java RocketMQ
MQ产品使用合集之在同一个 Java 进程内建立三个消费对象并设置三个消费者组订阅同一主题和标签的情况下,是否会发生其中一个消费者组无法接收到消息的现象
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
118 1
|
消息中间件 安全 Go
动态订阅时 rocketmq-client-go 代码有map并发bug
动态订阅时 rocketmq-client-go 代码有map并发bug
65 2
|
6月前
|
传感器 监控 网络协议
MQTT 发布、订阅模式介绍
【2月更文挑战第17天】
423 6
MQTT 发布、订阅模式介绍