Spring - Using @JmsListener to listen JMS messages

Spring supports annotation-driven listener endpoints. We need to use @JmsListener annotation on our listener methods.

Example

A MessageListener

package com.logicbig.example;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class MyJmsListener {
  @JmsListener(destination = AppConfig.QUEUE_NAME)
  public void handleMessage(String message) {//implicit message type conversion
      System.out.println("received: "+message);
  }
}

Using JmsTemplate to send messages

@Component
public class MyMessageSender {

  @Autowired
  private ConnectionFactory connectionFactory;
  private JmsTemplate jmsTemplate;

  @PostConstruct
  public void init() {
      this.jmsTemplate = new JmsTemplate(connectionFactory);
  }

  public void sendMessage(String message) {
      System.out.println("sending: " + message);
      jmsTemplate.send(AppConfig.QUEUE_NAME, new MessageCreator() {
          @Override
          public Message createMessage(Session session) throws JMSException {
              return session.createTextMessage(message);
          }
      });
  }
}

Java config and main class

We are going to use @EnableJms to enable JMS listener annotated endpoints. Also, we need to register DefaultJmsListenerContainerFactory instead of DefaultMessageListenerContainer (see last example) to enable this support.

@Configuration
@ComponentScan
@EnableJms
public class AppConfig {
  public static final String QUEUE_NAME = "example.queue";

  @Bean
  public ConnectionFactory connectionFactory() {
      ConnectionFactory connectionFactory =
              new ActiveMQConnectionFactory("vm://localhost");
      return connectionFactory;
  }

  @Bean
  public JmsListenerContainerFactory jmsListenerContainerFactory() {
      DefaultJmsListenerContainerFactory factory =
              new DefaultJmsListenerContainerFactory();
      factory.setConnectionFactory(connectionFactory());
      //core poll size=4 threads and max poll size 8 threads
      factory.setConcurrency("4-8");
      return factory;
  }

  public static void main(String[] args) throws InterruptedException {
      AnnotationConfigApplicationContext context =
              new AnnotationConfigApplicationContext(AppConfig.class);

      MyMessageSender ms = context.getBean(MyMessageSender.class);
      ms.sendMessage("test message 1");
      ms.sendMessage("test message 2");

      System.out.println("-- shutting down listener container --");
      JmsListenerEndpointRegistry bean = context.getBean(JmsListenerEndpointRegistry.class);
      for (MessageListenerContainer listenerContainer : bean.getListenerContainers()) {
          DefaultMessageListenerContainer container = (DefaultMessageListenerContainer) listenerContainer;
          container.shutdown();
      }
  }
}

Output

INFO | Using Persistence Adapter: KahaDBPersistenceAdapter[D:\LogicBig\example-projects\spring-jms\activemq-data\localhost\KahaDB]
 INFO | JMX consoles can connect to service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi
 INFO | KahaDB is version 6
 INFO | PListStore:[D:\LogicBig\example-projects\spring-jms\activemq-data\localhost\tmp_storage] started
 INFO | Apache ActiveMQ 5.15.0 (localhost, ID:JoeMsi-59330-1504928963493-0:1) is starting
 INFO | Apache ActiveMQ 5.15.0 (localhost, ID:JoeMsi-59330-1504928963493-0:1) started
 INFO | For help or more information please see: http://activemq.apache.org
 WARN | Store limit is 102400 mb (current store usage is 0 mb). The data directory: D:\LogicBig\example-projects\spring-jms\activemq-data\localhost\KahaDB only has 52369 mb of usable space. - resetting to maximum available disk space: 52369 mb
 INFO | Connector vm://localhost started
sending: test message 1
received: test message 1
sending: test message 2
received: test message 2
-- shutting down listener container --
 INFO | Connector vm://localhost stopped
 INFO | Apache ActiveMQ 5.15.0 (localhost, ID:xxx) is shutting down
 INFO | PListStore:[D:\LogicBig\example-projects\spring-jms\activemq-data\localhost\tmp_storage] stopped
 INFO | Stopping async queue tasks
 INFO | Stopping async topic tasks
 INFO | Stopped KahaDB
 INFO | Apache ActiveMQ 5.15.0 (localhost, ID:xxx) uptime 1.907 seconds
 INFO | Apache ActiveMQ 5.15.0 (localhost, ID:xxx) is shutdown

Example Project

Dependencies and Technologies Used:

• spring-context 4.3.10.RELEASE: Spring Context.
• spring-jms 4.3.10.RELEASE: Spring JMS.
• activemq-all 5.15.0: Puts together an ActiveMQ jar bundle.
• JDK 1.8
• Maven 3.3.9