Bad Queue consuming latency

[raymanf]

Hi,

I have integrated HorentQ-2.2.14.Final into Spring3.1 container.

I have  very extreme latency problems while using jms template(in order to send messages) and DMLC(message listener consumer) while using the point to point(queue) hornetq structure.

this is very important note: when I use topic(pub-sub mode) the latency is reasonable.

I did measures of latency after warming up (1000000 messages)

the scenario is simple:

Sending via jms-template a simple message to queue. and in the other side I have DMLC retrieve the message.

I calculate the message deliver's latency at 2 points:

before sending and after retrieving.

the latency being calculated in miliseconds and it's about 8 mili-seconds for one simple execution. that considers ridiculous alot for doing almost nothing.

*all components are in the same application and jvm

I am considering checking another jms provider to negative Hornetq's mechanisem's problem but before I do that I wanted to make sure with you guys mybe I have configured something worng or should I pay attention to somethign else?

I have spoke with Spring support and that was their answer:

Tuning for jms would require not only on the consumer and producer but as well on the jms provider. I am not familiar with HornetQ there is a grey area on some extent on how I would understand messages are handled. Depends on how messages are persistented in queue, tendency is there is an IO/disk overhead that you may have to consider.

One of our products is ActiveMq and I have dealt with cases on slow consumers. I know both activemq and hornetq behaves differently but usually here are our recommendation on tuning consumers, this may or may not work with hornetQ.

> Setting the prefetch value to a lower value - try starting with 1. You may want to check hornetQ further on how it handles pre fetch handling/policies

> Check behaviour of concurrency of your consumers. Check if activeConsumers are scaling or reaching to maxConcurrentConsumer depends on the performance of both application/jms provider/number of messages. MaxConcurrentConsumers might need to be increased. You may either do this by adding additional logging or exposing the listener attributes and monitor it via jmx.

> Separate a cacheconnectionFactory for consumers. Create two cacheconnectionFactory - One for the producer and the consumer. So that we can make sure the producers does not affect consumers in terms of waiting time for connection on the pool.

> Additional logging when sending. Since you are using your own message id on messages it might help if you can add logging before you send the message. The HornetQMessage.toString seems not to provide any info with regards to messages send. No way to check contents of a message before sending and when it was received currently it only prints it as null.

2013-03-10 13:48:41,199 org.springframework.jms.core.JmsTemplate [DEBUG] Sending created message: HornetQMessage[null]:PERSISTENT

I have attached my configuration files.

thanks for your help,

ray.

[ataylor]

its not really latency you are talking about it is throughput. You are sending persistant messages so throughput will be ultimately be limited by your disk I/O performance. If you were using non persistant messages then your throughput will be limited by your network speed.

Also make sure if you are using Spring that you use the cached connections, if not you will b ecreating a new connection for each message sent.

[raymanf]

Hi Andy,

"..its not really latency you are talking about it is throughput"

Why you assumed it's throughput I was talking about? I didnt measure how many messages per x time I am getting. I am measuring how fast they retrieved to thire dest. So when i execute 1 message(after warm up) It gets from one point to the second point within average of 8 miliseconds. if you take the same message and sending it within pub-sub mode it takes 1 miliseconds.

"..Also make sure if you are using Spring that you use the cached connections"

I am using cached connections for sure.

thanks,

ray.

[ataylor]

Why you assumed it's throughput I was talking about? I didnt measure how many messages per x time I am getting. I am measuring how fast they retrieved to thire dest. So when i execute 1 message(after warm up) It gets from one point to the second point within average of 8 miliseconds. if you take the same message and sending it within pub-sub mode it takes 1 miliseconds.

because you said "Sending via jms-template a simple message to queue. and in the other side I have DMLC retrieve the message." this is throughput, you didnt really give me any more info to go on

[raymanf]

So basiclly thats what I do. sending pretty light jms messaging containing timestamp of the first point.

and in the second point(on the message listener) I am extracting the first point timestamp and calculating the time diffreneces (currentTimestamp - firstPointTimeStamp).

the queue is integrated by HornetQ mechanizem.

I am on the same application in the same jvm.

[ataylor]

can you post your code for both pub/sub and pt to pt

[raymanf]

I can.. but it's mostly Xml Spring configuration:

Topic code:

Java:

   Producer:

private JmsTemplate rawPriceTopicJmsTemplate;
rawPriceTopicJmsTemplate.send(new MessageCreator()
                              {
                                        @Override
                                        public Message createMessage(Session session) throws JMSException
                                        {

                                                  TextMessage message = session.createTextMessage("some msg");
                                                  long currentTimeMillis = System.currentTimeMillis();
                                                  message.setIntProperty("counter", counter++);
                                                  message.setLongProperty("timeStamp", currentTimeMillis);
                                                  return message;
                                        }
                              });

Consumer: (I am using here my own ExecuterService only within the consumer's topic mechanizem)

public class ClientFeedTopicMDBBeanImpl implements MessageListener
{
          static Logger logger = Logger.getLogger(ClientFeedTopicMDBBeanImpl.class);
          private ExecutorService threadPool = Executors.newFixedThreadPool(100);;


          @Override
          public void onMessage(final Message msg)
          {
                    Runnable t = new Runnable()
                    {
                              public void run()
                              {
                                        executeMsg(msg);
                              }
                    };
                    threadPool.execute(t);
          }


          private void executeMsg(final Message msg)
          {
                    try
                    {


                              long start = msg.getLongProperty("timeStamp");
                              long end = System.currentTimeMillis();
                              long latency = end - start;
                              logger.debug("got msg=" + msg.getIntProperty("counter") + ", latency=" + latency);
                    }
                    catch (JMSException e)
                    {           
                              e.printStackTrace();
                    }
          }
}

Spring: JMSTemplate and DLMC configuration:

<bean id="clientFeedTopicJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
                    <property name="connectionFactory" ref="cachedConnectionFactory" />
                    <property name="defaultDestination" ref="clientFeedTopic" />
                    <property name="pubSubDomain" value="true" />
                    <property name="receiveTimeout" value="3000" />
                    <property name="messageIdEnabled" value="false" />
                    <property name="messageTimestampEnabled" value="true" />
          </bean>

<bean id="clientFeedTopsicDMLC"
                    class="org.springframework.jms.listener.DefaultMessageListenerContainer">
                    <property name="connectionFactory" ref="cachedConnectionFactory" />
                    <property name="pubSubDomain" value="true" />
                    <property name="concurrentConsumers" value="1" />
                    <property name="destination" ref="clientFeedTopic" />
                    <property name="messageListener" ref="clientFeedTopicMDBBean" />
                    <property name="recoveryInterval" value="10000" />
          </bean>

Queue:

   Java - producer java remaines the same.

            consumer:

public class PricesQueueMDBImpl implements MessageListener, PricesQueueMDB
{
          static Logger logger = Logger.getLogger(PricesQueueMDBImpl .class);


          @Override
          public void onMessage(final Message msg)
          {
                    try
                    {


                              long latency = System.currentTimeMillis() - msg.getLongProperty("timeStamp");
                              logger.debug("got msg=" + msg.getIntProperty("counter") + ",           latency=" + latency);


                    }
                    catch (JMSException e)
                    {

                              e.printStackTrace();
                    }
          }

Spring: (this time pubSubDomain value will be false - means point to point(Queue)

<bean id="PricesQueueJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
                    <property name="connectionFactory" ref="cachedConnectionFactory" />
                    <property name="defaultDestination" ref="pricesQueue" />
                    <property name="pubSubDomain" value="false" />
                    <property name="receiveTimeout" value="3000" />
                    <property name="messageIdEnabled" value="false" />
                    <property name="messageTimestampEnabled" value="true" />
          </bean>

<bean id="PricesQueueDMLC"
                    class="org.springframework.jms.listener.DefaultMessageListenerContainer">
                    <property name="connectionFactory" ref="cachedConnectionFactory" />
                    <property name="pubSubDomain" value="false" />
                    <property name="concurrency" value="10-50" />
                    <property name="destination" ref="pricesQueue" />
                    <property name="messageListener" ref="PricesQueueMDBBean" />
                    <property name="recoveryInterval" value="10000" />
          </bean> 

This is hornetQ's configuration in xml:

<bean id="cachedConnectionFactory"

                    class="org.springframework.jms.connection.CachingConnectionFactory"

                    p:targetConnectionFactory-ref="connectionFactory" p:sessionCacheSize="200" />

          <bean name="fileConfiguration" class="org.hornetq.core.config.impl.FileConfiguration"

                    init-method="start" destroy-method="stop" />

          <bean name="hornetQSecurityManagerImpl"

                    class="org.hornetq.spi.core.security.HornetQSecurityManagerImpl" />

          <!-- The core server -->

          <bean name="hornetQServerImpl" class="org.hornetq.core.server.impl.HornetQServerImpl">

                    <constructor-arg ref="fileConfiguration" />

                    <constructor-arg ref="mbeanServer" />

                    <constructor-arg ref="hornetQSecurityManagerImpl" />

          </bean>

          <!-- The JMS server -->

          <bean name="jmsServerManagerImpl" class="org.hornetq.jms.server.impl.JMSServerManagerImpl"

                    init-method="start" destroy-method="stop" depends-on="namingServer">

                    <constructor-arg ref="hornetQServerImpl" />

          </bean>

          <bean name="connectionFactory" class="org.hornetq.jms.client.HornetQJMSConnectionFactory">

                    <constructor-arg name="ha" value="true" />

                    <constructor-arg>

                              <bean class="org.hornetq.api.core.TransportConfiguration">

                                        <constructor-arg

                                                  value="org.hornetq.core.remoting.impl.netty.NettyConnectorFactory" />

                                        <constructor-arg>

                                                  <map key-type="java.lang.String" value-type="java.lang.Object">

                                                            <entry key="port" value="5445"></entry>

                                                  </map>

                                        </constructor-arg>

                              </bean>

                    </constructor-arg>

          </bean>

Hope this will be enough. please let me know if you need anything else.

thanks,

ray.

[ataylor]

ok, thats all spring code I was hoping to see how the underlying consumers were created so i can't really comment. Is your subscriber durable, because if its not you are basically just sending to a non durable queue, whcih would mean the message isnt persisted.

[clebert.suconic]

It's not clear to me how you're sending a message...

But it seems you are measuring a lot of stuff there... From the time you created the producer, the session, persisted and still consumed the message.

You need to optimize your code to just receive messages (not any other crap).. and only measure after an initial period (after some messages were sent).  usually sprint template will create a consumer and close it for every message you receive.. unless you do some optimization about it.

If you summarize it to a simple hornetq example we comment better.. I honestly don't have any cycles to drill down at spring's code at the moment.. I wish I did.. but I don't.

[clebert.suconic]

Things we usually look for.. when you create a session, is it persisted, transactional? when you create a consumer (messageConsumer)....  when you ack.?

[raymanf]

"But it seems you are measuring a lot of stuff there... From the time you created the producer, the session, persisted and still consumed the message..."

I only measure the latency between the time of execution and the received time.

                            long start = msg.getLongProperty("timeStamp"); // the executed time
                              long end = System.currentTimeMillis(); //the current time
                              long latency = end - start; //the diff between them
                              logger.debug("got msg=" + msg.getIntProperty("counter") + ", latency=" + latency);

I do the same logic  while measuring Topic and it was like 1ms. with queue it's around 7-8ms

[raymanf]

Things we usually look for.. when you create a session, is it persisted, transactional? when you create a consumer (messageConsumer)....  when you ack.?

Re: Bad Queue consuming latency

it's not transactional. The consumer is created once and the session is cached. I guess your lack of Spring support with all concern to DMLC/JMStemplate makes it hard to support this case.

[ataylor]

what clebert is saying is that unless you tell us how you are creating your consumers and sessions we cant help you, this has nothing to do with lack of spring support at all, in fact there is nothing for us to support. What you need to do is gain an understanding of what Spring is doing under the covers and how to configure it. For instance, i asked you if your  subscriptions were durable or not and you never answered.

[clebert.suconic]

Andy is right... besides you are measuring probably create consumer, creating session and creating producer times...  you need to isolate the meausres.

And BTW: What Spring support has nothing to do on how you properly measure micro benchmarks.  We won't go anywhere if you don't define the measurement correctly.. and if we start drilling down on your implementation that will measure your architecture as opposed to the messaging infra structure.

It's impossible for us to get down to what you are doing without proper context. So, you need to define your use case better or it will be nearly impossible to anyone to help you.

[ataylor]

I only measure the latency between the time of execution and the received time.

Clebert is correct here as well, your not really measuring anything, and to measure 'latency' between a producer and a client makes no sense. messaging systems are asynchronous, the producer and consumer aren't connected to each other. When you talk about latency you are typically talking about network round trip time, so measuring the latency of a producer sending a message and receiving an acknowledgement makes sense or measuring the time for a server to send a message to a consumer and receive an ack. When measuring performance on messaging systems you measure throughput and then test this by scaling both horizontally and vertically. Your test doesnt even measure when the consumer receives the message only when your onMessage is called, there may have been 100 messages in the clients buffer, garbage collection may have happened, thread context switching, etc etc. To sum up if you want to measure the performance send millions of messages between lots of clients and destinations and measure the throughput.