Eclipse Community Forums: Paho » Concurrent subscribe, unsubscribe and publish operations using same MqttClient instance

Home » Eclipse Projects » Paho » Concurrent subscribe, unsubscribe and publish operations using same MqttClient instance(MqttClient times out when there are concurrent subscribe, unsubscribe and publish operations on same MqttClient instance)

Concurrent subscribe, unsubscribe and publish operations using same MqttClient instance [message #1727001]

Fri, 18 March 2016 03:50

Eclipse User

Hi,

I am using Paho Mqtt V3 (version 1.0.2) in my application which is running in production now. My application is a rest based application running in Tomcat and connects to ActiveMQ broker (which is configured to use mqtt+nio as transport connector). The app creates 6-7 instances of MqttClient in the same JVM and each MqttClient instance performs concurrent subscribe/unsubscribe/publish operations.

Recently we have been seeing many instances of the following exception in our logs :

Timed out waiting for a response from the server (32000)
at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:31)
at org.eclipse.paho.client.mqttv3.internal.ClientState.checkForActivity(ClientState.java:564)
at org.eclipse.paho.client.mqttv3.internal.ClientComms.checkForActivity(ClientComms.java:660)
at org.eclipse.paho.client.mqttv3.TimerPingSender$PingTask.run(TimerPingSender.java:77)
at java.util.TimerThread.mainLoop(Timer.java:555)
at java.util.TimerThread.run(Timer.java:505)

To reproduce this issue, I just wrote a stand-alone Java class which creates 5 instances of MqttClient in 5 threads. Each thread then spawns further 5 threads all of which use the same MqttClient instance to perform subscribe/publish operations continuously. After running this for about 5-15 minutes, I get the "Timed out" exception. However, there are no errors/warnings in the ActiveMQ logs.

Is my usage of MqttClient wrong in any way?

I have attached the sample concurrent publisher and subscriber classes that can be used to reproduce the problem.

Any help or pointers would be greatly appreciated.

Thanks,
Shobhana

Attachment: ConcurrentTopicCreator.java
(Size: 3.47KB, Downloaded 266 times)
Attachment: ConcurrentPublisher.java
(Size: 3.50KB, Downloaded 281 times)

Re: Concurrent subscribe, unsubscribe and publish operations using same MqttClient instance [message #1727913 is a reply to message #1727001]

Tue, 29 March 2016 06:49

Eclipse User

Hi, It could be that you are maxing out the number of inflight messages by flooding it with so many publishes at the same time. Could you restructure your application to have an internal queue that you push messages to which an MQTT client then publishes from on it's own thread?

Re: Concurrent subscribe, unsubscribe and publish operations using same MqttClient instance [message #1727915 is a reply to message #1727913]

Tue, 29 March 2016 07:06

Eclipse User

Thanks for your reply James.

When the number of inflight messages are more, we get a different exception (Too many publishes in progress) and to avoid this, I have already changed my application to use a queue mechanism just as you explained. But even after this change, I see a lot of "Timed out" exceptions.

To check if my (poor?) configuration of ActiveMQ was causing this problem, I tested with a stand-alone JMS test client (using only ActiveMQ client jar and replaced MQTT publisher and subscriber with JMS publisher and subscriber) but could not reproduce this problem.

Can you think of any other reason?

Re: Concurrent subscribe, unsubscribe and publish operations using same MqttClient instance [message #1731504 is a reply to message #1727915]

Fri, 06 May 2016 06:05

Eclipse User

I would try using a different MQTT Broker instead of a different connector type, as by switching to a JMS connector you'll no longer be hitting any MQTT code at all on the ActiveMQ side. Try with something like Mosquitto, if you see the same results then it might be worth making sure that if you are listening for any callbacks that you are not holding them up unnecessarily before releasing them as that can cause time outs in certain cases.

Previous Topic:	Paho MQTT C Client was hung-up while connecting to an non-existing Broker
Next Topic:	Publish message from callback messageArrived()

Goto Forum:

-=] Back to Top [=-

Current Time: Thu Nov 06 03:36:11 EST 2025

.:: Contact :: Home ::.

Breadcrumbs

Sign up to our Newsletter