CODEWITHSUNDEEP
activemq-classicproducer-consumer
Ali Çevik
Ali Çevik

Reputation: 3

Stop sending messages to broker

In my scenario I have one boroker, one producer, one consumer.I am using activemq for writing my app. logs to db.As u know writing logs to db is time taking process.That's why consumer is more and more slow than producer.For ex. I send 100.000 message(huge objects).Producer finishes sending messages in 20 mins.But When the producer finished, consumer has finished 4.000 message processing yet.

My question is ; is there any way to say procuder "wait if there are x messages in the broker waiting for consuming and try to send after a while"

how can solve this problem.

Upvotes: 0

Views: 483

Answers (2)

Jakub Korab
Jakub Korab

Reputation: 5034

The idea behind using a piece of middleware like ActiveMQ is to seperate production and consumption of messages over time. For example, if the consuming system is down, your producer shouldn't care - all it cares about is that the messages went into the queue. Once there ActiveMQ will delivered these messages at a later point in time.

If you are having trouble consuming messages fast enough, you would be far better off to increase the number of consumers or speeding up the processing of the message once received.

Upvotes: 1

Mayoares
Mayoares

Reputation: 1284

There are several ways to monitor ActiveMQ.

For example you could use Advisory-Messages. There are some indicators available like

ActiveMQ.Advisory.SlowConsumer.Queue
ActiveMQ.Advisory.FastProducer.Topic

etc.

To find out the queue size exactly per JMX, please read http://activemq.apache.org/how-do-i-find-the-size-of-a-queue.html

and try for example something similar as my code snippet: 

JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://<server>:<port>/jmxrmi");
JMXConnector jmxConnector = JMXConnectorFactory.connect(url);
...
try
{
  jmxConnector.connect();
  MBeanServerConnection connection = jmxConnector.getMBeanServerConnection();
...
  BrokerViewMBean brokerBean = MBeanServerInvocationHandler.newProxyInstance(connection, brokerBeanName, BrokerViewMBean.class, true);
  System.out.println("Total message count:" + brokerBean.getTotalMessageCount());
...
}
...

If you monitor the QueueSize/TotalMessageCount, you could react on that.

Upvotes: 1

Related Questions