How to listen to the message queue from a web application? (Tomcat, ActiveMQ)

Question

I am happily improving my web application that runs on Apache Tomcat . To send and receive messages, the JMS ActiveMQ server has been added.

I can already send and receive messages, but you need help on the recipient side.

How will my web application constantly listen to one queue for receiving messages?

New messages arrive, and the server must act on them. Example: adding data to the database or sending a message back.

I can already send messages. This is the code.

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); Connection connection = factory.createConnection(); Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE); Queue queue = session.createQueue("clientQueue"); MessageProducer publisher = session.createProducer(queue); connection.start(); Message message = null; message = session.createTextMessage("Text Message"); publisher.send(message); 

I can already receive the message after the request (click ;-))

 connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616"); connection = connectionFactory.createConnection(); connection.start(); session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE); destination = session.createQueue("serverQueue"); consumer = session.createConsumer(destination); while (true) { Message message = consumer.receive(300000); //Do message stuff } 

How can I let a web application listen to the queue continuously? What is the recommended way?

All help is appreciated. Thanks.

EDITING - SOLUTION

Current working solution with DaveH offers

I added a ServletContextListener to listen to my message continuously.

web.xml

 <listener> <listener-class>com.test.JMSContextListener</listener-class> </listener> 

Listeren:

 public class JMSContextListener implements ServletContextListener { @Override public void contextInitialized(ServletContextEvent arg0) { Thread thread = new Thread(new JMSConnector()); thread.start(); } @Override public void contextDestroyed(ServletContextEvent arg0) { //Nothing } } 

Compound:

 public class JMSConnector implements Runnable { public void run() { try { Context context = new InitialContext(); QueueConnectionFactory factory = (QueueConnectionFactory) context.lookup("java:comp/env/jms/ConnectionFactory"); Connection connection = factory.createConnection(); Queue queue = (javax.jms.Queue) context.lookup("java:comp/env/jms/serverQueue"); Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE); MessageConsumer consumer = session.createConsumer(queue); //This MessageListener will do stuff with the message MessageListenerImpl messageListener = new MessageListenerImpl(); consumer.setMessageListener(messageListener); connection.start(); // Start connection or nothing will happen!!! connection.start(); } catch (JMSException ex) { //TODO } catch (NamingException ex) { //TODO } } } 

Is this the recommended way or should it be improved?

All help is appreciated. Thanks.

Answer 1

If you have code that can already consume messages from the queue (which apparently you are doing), I think your problem boils down to how you can run this piece of code.

It seems that you are not using any frameworks, so I think the approach I would take is to take the code that can extract messages from the queue and run it in a separate thread on the application server. Get this thread to get started from the moment the application server starts, and tidy up when the application server closes.

The easiest way to start the thread when the application server starts is to enter the ServletContextListener (see the example here .) In context listening, start the queue by listening to the code in a separate thread.

EDIT: I used this proposed solution and added code to the question.

Answer 2

I use spring to listen on the queue. listener definition looks like this:

 <jms:listener-container connection-factory="jmsConnectionFactoryLocal"> <jms:listener destination="QUEUE_NAME" ref="channelManagerSimulatorDefault"/> </jms:listener-container> 

jmsConnectionFactoryLocal should be created according to your MQ. in my case it was IBM WebsphereMQ, so the jmsConnectionFactoryLocal definition is as follows:

 <bean id="mqConnectionFactoryLocal" class="com.ibm.mq.jms.MQQueueConnectionFactory"> <property name="hostName"> <value>THE_MQ_SERVER_IP</value> </property> <property name="port"> <value>MQ_PORT</value> </property> <property name="queueManager"> <value>QUEUE_MANAGER_NAME</value> </property> <property name="transportType"> <value>1</value> </property> </bean> <bean id="jmsConnectionFactoryLocal" class="org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapter"> <property name="targetConnectionFactory" ref="mqConnectionFactoryLocal"/> <property name="username" value="USER_NAME"/> <property name="password" value="PASSWORD"/> </bean> 

you need to find the correct ConnectionFactory implementation for ActiveMQ and use it. the listener and jmsConnectionFactory are the same and independent of the MQ provider.

Answer 3

I am using activeMQ in my webApplication spring -mvc and JMS Template with the following approach.

 <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL"> <value>tcp://localhost:61616</value> </property> </bean> <bean id="messageSender" class="xyz.MessageSender"/> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="connectionFactory" /> <property name="defaultDestination" ref="destination" /> </bean> <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg value="REQUEST_QUEUE" /> </bean> <bean id="messageListener" class="xtz.MessageListener"> <property name="listenerId" value="1" /> </bean> <jms:listener-container connection-factory="connectionFactory"> <jms:listener destination="RESPONSE_QUEUE" ref="messageListener" method="messageReceived" /> </jms:listener-container> 

The sender and listener class implementation given below.

 public class MessageSender { @Autowired private JmsTemplate jmsTemplate; public void sendMessage() { jmsTemplate.send(new MessageCreator() { public Message createMessage(Session session) throws JMSException { MapMessage message = session.createMapMessage(); message.setString("messageType", XXX); message.setString("jsonMessage", XXXX); return message; } }); } } public class MessageListener { private int listenerId; @Override public void messageReceived(Map<String, Object> message) throws Exception { //put your logic here } public int getListenerId() { return listenerId; } public void setListenerId(int listenerId) { this.listenerId = listenerId; } } 

Answer 4

• Setting up a JMS queue in Tomcat: https://martinsdeveloperworld.wordpress.com/2013/03/03/apache-activemq-and-tomcat/

• Put activemq-all-5.xx.jar in $ TOMCAT_HOME / lib

• Listening to a loop in a thread started in @WebListener :

• Stop when contextDestroyed

   @WebListener @Slf4j public class JmsMailListener implements ServletContextListener { private Thread listenerThread = null; private QueueConnection connection; @Override public void contextInitialized(ServletContextEvent sce) { try { InitialContext initCtx = new InitialContext(); ActiveMQConnectionFactory connectionFactory = (ActiveMQConnectionFactory) initCtx.lookup("java:comp/env/jms/ConnectionFactory"); connectionFactory.setTrustAllPackages(true); connection = connectionFactory.createQueueConnection(); QueueSession queueSession = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue = (Queue) initCtx.lookup("java:comp/env/jms/queue/MailQueue"); QueueReceiver receiver = queueSession.createReceiver(queue); connection.start(); log.info("Listen JMS messages ..."); listenerThread = new Thread(() -> { try { while (!Thread.interrupted()) { Message m = receiver.receive(); if (m instanceof ObjectMessage) { ObjectMessage om = (ObjectMessage) m; MyObject myObject = (MyObject) om.getObject(); log.info("Received MyObject {}", myObject); ... } } } catch (Exception e) { log.error("Receiving messages failed: " + e.getMessage(), e); } }); listenerThread.start(); } catch (Exception e) { log.error("JMS failed: " + e.getMessage(), e); } } @Override public void contextDestroyed(ServletContextEvent sce) { if (connection != null) { try { connection.close(); } catch (JMSException ex) { log.warn("Couldn't close JMSConnection: ", ex); } } if (listenerThread != null) { listenerThread.interrupt(); } } }