This tutorial will show you how we can integrate Spring and Apache ActiveMQ using point-to-point messaging domain. In point to point communication there are exactly one message producer and one message consumer. I will show you both XML configuration and annotation based configuration using Spring JMS integration.
For more information on point-to-point messaging system please read tutorial https://roytuts.com/configure-jms-client-using-glassfish-3/
Now we will look into the following steps in order to implement point-to-point messaging system using Spring and ActiveMQ integration.
Prerequisites
Read tutorial https://roytuts.com/apache-activemq-configuration-in-windows/ for configuring ActiveMQ but you do not need to create any Queue.
Eclipse 4.12, Java 8 or 12, ActiveMQ 5.15.10, Spring Boot 2.2.1, Spring 4.1.5, Maven 3.6.1, Gradle 5.6
Creating Project
For Spring Boot we will create gradle based project in Eclipse. The name of the project is spring-jms-activemq-point-to-point.
Updating Build Script
Update the gradle.build script to include the required dependencies for application.
buildscript {
ext {
springBootVersion = '2.2.1.RELEASE'
}
repositories {
mavenCentral()
}
dependencies {
classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
}
}
plugins {
id 'java-library'
id 'org.springframework.boot' version '2.2.1.RELEASE'
}
sourceCompatibility = 12
targetCompatibility = 12
repositories {
mavenCentral()
}
dependencies {
implementation("org.springframework.boot:spring-boot-starter-activemq:${springBootVersion}")
implementation('org.apache.activemq:activemq-broker:5.15.11')
}If you are using maven based project, then using Spring version 4.1.5 and Java 8, you can use below pom.xml file or you can upgrade the Spring and Java versions:
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.roytuts</groupId>
<artifactId>spring-jms-activemq-publish-subscribe</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<jdk.version>1.8</jdk.version>
<junit.version>4.11</junit.version>
<slf4j.version>1.7.5</slf4j.version>
<activemq.version>5.11.1</activemq.version>
<spring.version>4.1.5.RELEASE</spring.version>
</properties>
<dependencies>
<!-- activemq -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>${activemq.version}</version>
</dependency>
<!-- Spring -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${spring.version}</version>
</dependency>
<!-- junit -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>${junit.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>${jdk.version}</source>
<target>${jdk.version}</target>
</configuration>
</plugin>
</plugins>
</build>
</project>Property File
For Spring Boot application we will create application.properties file under classpath directory src/main/resources with the following content:
JMS.BROKER.URL=tcp://localhost:61616
JMS.QUEUE.NAME=IN_QUEUEFor XML based configuration we will create activemq-jms-spring-properties.xml under classpath directory src/main/resources.
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context"
xmlns:jee="http://www.springframework.org/schema/jee" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-3.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
<bean id="jmsSpringProperties"
class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"
name="jmsSpringProperties">
<property name="order" value="99999" />
<property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE" />
<property name="ignoreUnresolvablePlaceholders" value="true" />
<property name="properties">
<value>
<!-- JMS -->
JMS.BROKER.URL=tcp://localhost:61616
JMS.QUEUE.NAME=IN_QUEUE
</value>
</property>
</bean>
</beans>The property file just declares ActiveMQ broker URL and queue name as key/value pairs that will be used for messaging system.
Message Producer
Create a class called MessageProducer that will produce message or send message to the destination – Queue.
We are using here JmsTemplate API to send the message to the queue provided by Spring framework.
Please note you don’t need @Component annotation for XML based Spring application.
package com.roytuts.spring.jms.activemq.point.to.point.producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;
@Component
public class MessageProducer {
@Autowired
private JmsTemplate jmsTemplate;
public void sendMessageToDefaultDestination(final String message) {
jmsTemplate.convertAndSend(message);
}
}Message Consumer
Create a class called MessageDefaultConsumer that will receive message from the destination – Queue.
We are just accepting text message from the queue and for all other message it will throw an exception.
We mark consumer with @Component annotation to avoid creating bean ourselves.
Please note you don’t need @Component annotation for XML based Spring application.
package com.roytuts.spring.jms.activemq.point.to.point.consumer;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import org.springframework.stereotype.Component;
@Component
public class MessageConsumer implements MessageListener {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
String msg = ((TextMessage) message).getText();
System.out.println("Message has been consumed : " + msg);
} catch (JMSException ex) {
throw new RuntimeException(ex);
}
} else {
throw new IllegalArgumentException("Message must be of type TextMessage");
}
}
}Create Configuration
Next we need to configure our application to create various beans such as Connection Factory, Queue, Message Container Factory etc.
We will tell you both XML based and annotation based configurations.
Create an XML configuration file called activemq-spring-jms.xml that contains JMS related configuration for the application. Put this file under src/main/resources/jms.
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">
<!-- ActiveMQ connection factory -->
<bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<constructor-arg index="0" value="${JMS.BROKER.URL}" />
</bean>
<!-- ConnectionFactory Definition -->
<bean id="connectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<constructor-arg ref="amqConnectionFactory" />
</bean>
<!-- Destination Queue -->
<bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="${JMS.QUEUE.NAME}" />
</bean>
<!-- JmsTemplate Definition -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="destinationQueue" />
</bean>
<!-- Message Producer -->
<bean id="messageProducer" class="com.roytuts.spring.jms.activemq.point.to.point.producer.MessageProducer" />
<!-- Message Consumer from Default Destination -->
<bean id="messageDefaultConsumer" class="com.roytuts.spring.jms.activemq.point.to.point.consumer.MessageConsumer" />
<!-- Message Consumer Container for Default Destination -->
<bean
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destinationName" value="${JMS.QUEUE.NAME}" />
<property name="messageListener" ref="messageDefaultConsumer" />
</bean>
</beans>For Spring Boot application create below config class:
package com.roytuts.spring.jms.activemq.point.to.point.config;
import javax.jms.ConnectionFactory;
import javax.jms.Queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.core.env.Environment;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.MessageListenerContainer;
import com.roytuts.spring.jms.activemq.point.to.point.consumer.MessageConsumer;
@Configuration
@PropertySource(value = "classpath:application.properties") // optional
public class JmsConfig {
@Autowired
private Environment env;
@Autowired
private MessageConsumer messageConsumer;
@Bean
public ConnectionFactory connectionFactory() {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(env.getProperty("JMS.BROKER.URL"));
return connectionFactory;
}
@Bean
public CachingConnectionFactory cachingConnectionFactory() {
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory());
return cachingConnectionFactory;
}
@Bean
public Queue queue() {
Queue queue = new ActiveMQQueue(env.getProperty("JMS.QUEUE.NAME"));
return queue;
}
@Bean
public JmsTemplate jmsTemplate() {
JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory());
jmsTemplate.setDefaultDestination(queue());
return jmsTemplate;
}
@Bean
public MessageListenerContainer messageListenerContainer() {
DefaultMessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer();
messageListenerContainer.setDestination(queue());
messageListenerContainer.setMessageListener(messageConsumer);
messageListenerContainer.setConnectionFactory(connectionFactory());
return messageListenerContainer;
}
}or XML configuration, we will load all XML configuration into one XML file. Therefore, create an XML file called activemq-jms-spring-context.xml that will load all other resources and configure support for annotation in the application. Put this file under src/main/resources/spring.
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:annotation-config></context:annotation-config>
<import resource="classpath:activemq-jms-spring-properties.xml" />
<import resource="classpath:jms/activemq-spring-jms.xml" />
</beans>Create Main Class
We will create two separate main classes – one for Spring XML based configuration and another for Spring Boot application.
For XML based configuration, use below class:
public class SpringJmsActiveMqPointToPointApp {
public static void main(String[] args) {
ApplicationContext applicationContext = new ClassPathXmlApplicationContext(
"classpath:spring/activemq-jms-spring-context.xml");
MessageProducer messageProducer = (MessageProducer) applicationContext
.getBean("messageProducer");
messageProducer
.sendMessageToDefaultDestination("Send this message to default destination.");
}
}For Spring Boot application, use below class:
package com.roytuts.spring.jms.activemq.point.to.point;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import com.roytuts.spring.jms.activemq.point.to.point.producer.MessageProducer;
@SpringBootApplication(scanBasePackages = "com.roytuts.spring.jms.activemq.point.to.point")
public class SpringJmsActiveMqPointToPointApp implements CommandLineRunner {
@Autowired
private MessageProducer messageProducer;
public static void main(String[] args) {
SpringApplication.run(SpringJmsActiveMqPointToPointApp.class, args);
}
@Override
public void run(String... args) throws Exception {
messageProducer.sendMessageToDefaultDestination("Send this message to default destination.");
}
}Testing the Application
Running the main class you will get the following output in the console:
Message has been consumed : Send this message to default destination.Verify in ActiveMQ Web Console
Click on Queues in ActiveMQ Web Console. You will not see any message in the IN_QUEUE because as soon as you ran the MessageProducer the message has been consumed by the MessageConsumer and now you will see one message has been dequeued.
We do not need to run the MessageConsumer because it is an asynchronous messaging system and MessageConsumer is already registered to the Spring’s DefaultMessageListenerContainer.
So when a message arrives to the Queue, the message gets automatically consumed by the consumer’s onMessage() method.
Source Code
That’s all. Thank you for reading.