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)

Show: Today's Messages :: Show Polls :: Message Navigator

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

Fri, 18 March 2016 07:50

Shobhana Sriram

Messages: 2
Registered: March 2016

Junior Member

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 230 times)
Attachment: ConcurrentPublisher.java
(Size: 3.50KB, Downloaded 246 times)

Report message to a moderator

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

Tue, 29 March 2016 10:49

James Sutton

Messages: 71
Registered: July 2015

Member

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?

Report message to a moderator

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

Tue, 29 March 2016 11:06

Shobhana Sriram

Messages: 2
Registered: March 2016

Junior Member

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?

Report message to a moderator

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

Fri, 06 May 2016 10:05

James Sutton

Messages: 71
Registered: July 2015

Member

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.

Report message to a moderator

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 [=-

[ Syndicate this forum (XML) ] [

]

Current Time: Thu Apr 18 07:13:54 GMT 2024

.:: Contact :: Home ::.

Breadcrumbs

Sign up to our Newsletter