Back to Spring Boot Course

ActiveMQ is a widely used message broker developed by Apache that facilitates the communication between different parts of an application or between different applications through messaging. It supports the Java Message Service (JMS) and provides features like message persistence, transaction management, and high availability.

Key Concepts and Features

Message Broker:

• Role: Acts as an intermediary for sending and receiving messages between producers and consumers.
• Types: Supports both point-to-point (queues) and publish-subscribe (topics) messaging models.

JMS (Java Message Service):

• JMS API: A Java API that allows applications to create, send, receive, and read messages.
• Support: ActiveMQ fully supports JMS 1.1 and provides partial support for JMS 2.0.

Broker Modes:

• Embedded: Runs within the same JVM as the application.
• Standalone: Runs as a separate server process.
• Clustered: Multiple brokers work together to provide high availability and scalability.

Persistence:

• Message Stores: ActiveMQ supports different types of message stores such as in-memory, file-based, and database-backed stores.
• Durable Subscriptions: Ensures that messages are retained for durable subscribers even if they are not connected.

Transport Protocols:

• OpenWire: Native protocol used by ActiveMQ.
• AMQP (Advanced Message Queuing Protocol): Open standard protocol.
• STOMP (Simple Text Oriented Messaging Protocol): Simple text-based protocol.
• MQTT (Message Queuing Telemetry Transport): Lightweight protocol for small devices.

Configuration:

• XML: ActiveMQ configuration is primarily done using XML files.
• Spring Boot Integration: Spring Boot provides easy integration with ActiveMQ using properties files and auto-configuration.

Let's develop a Spring Boot application where we publish a message to ActiveMQ. A listener will then receive the message from ActiveMQ and store it in MongoDB.

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>3.4.0</version>
		<relativePath /> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.praveen</groupId>
	<artifactId>praveen-activemq-mongodb</artifactId>
	<version>1.0.0</version>
	<name>praveen-activemq-mongodb</name>
	<description>Praveen Active MQ and Mongo DB</description>
	<url />
	<licenses>
		<license />
	</licenses>
	<developers>
		<developer />
	</developers>
	<scm>
		<connection />
		<developerConnection />
		<tag />
		<url />
	</scm>
	<properties>
		<java.version>17</java.version>
	</properties>
	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-activemq</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-mongodb</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springdoc</groupId>
			<artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
			<version>2.7.0</version>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-configuration-processor</artifactId>
			<optional>true</optional>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>

</project>

application.properties

spring.application.name=praveen-activemq-mongodb
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin
spring.data.mongodb.uri=mongodb+srv://praveenoruganti:xxxxxxxx@praveenoruganti.neyvmhh.mongodb.net/testdb?retryWrites=true&w=majority&appName=praveenoruganti
app.queue-name=testQueue
app.collection-name=messages
app.database-name=testdb

ApplicationConfig.java

package com.praveen.config;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConfigurationProperties(prefix = "app")
public class ApplicationConfig {
	private String queueName;
	private String collectionName;
	private String databaseName;

	public String getQueueName() {
		return queueName;
	}

	public void setQueueName(String queueName) {
		this.queueName = queueName;
	}

	public String getCollectionName() {
		return collectionName;
	}

	public void setCollectionName(String collectionName) {
		this.collectionName = collectionName;
	}

	public String getDatabaseName() {
		return databaseName;
	}

	public void setDatabaseName(String databaseName) {
		this.databaseName = databaseName;
	}

}

Message.java

package com.praveen.bo;

public record Message(String content, int id) {
}

ActivemqMongodbApplication.java

package com.praveen;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class ActivemqMongodbApplication {

	public static void main(String[] args) {
		SpringApplication.run(ActivemqMongodbApplication.class, args);
	}

}

OpenAPIConfig.java

package com.praveen.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import io.swagger.v3.oas.models.Components;
import io.swagger.v3.oas.models.OpenAPI;
import io.swagger.v3.oas.models.info.Info;

@Configuration
public class OpenAPIConfig {

	@Bean
	OpenAPI customOpenAPI() {
		return new OpenAPI().components(new Components()).info(
				new Info().title("Active MQ and Mongo DB Test App").description("Active MQ and Mongo DB Test App"));
	}

}

MessageProducer.java

package com.praveen.queue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.praveen.bo.Message;
import com.praveen.config.ApplicationConfig;

@Service
public class MessageProducer {

	private static final Logger log = LoggerFactory.getLogger(MessageProducer.class);

	private final JmsTemplate jmsTemplate;
	private final ApplicationConfig applicationConfig;
	private final ObjectMapper objectMapper;

	public MessageProducer(JmsTemplate jmsTemplate, ApplicationConfig applicationConfig, ObjectMapper objectMapper) {
		super();
		this.jmsTemplate = jmsTemplate;
		this.applicationConfig = applicationConfig;
		this.objectMapper = objectMapper;
	}

	public void sendMessage(Message message) {
		try {
			jmsTemplate.convertAndSend(applicationConfig.getQueueName(), objectMapper.writeValueAsString(message));
		} catch (JsonProcessingException e) {
			log.error("Json Processing Exception {}", e.getMessage());
		}
	}
}

MessageListener.java

package com.praveen.queue;

import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.praveen.bo.Message;
import com.praveen.config.ApplicationConfig;

@Component
public class MessageListener {

	private static final Logger log = LoggerFactory.getLogger(MessageListener.class);

	private final ApplicationConfig applicationConfig;
	private final MongoClient mongoClient;
	private final ObjectMapper objectMapper;

	public MessageListener(ApplicationConfig applicationConfig, MongoClient mongoClient, ObjectMapper objectMapper) {
		super();
		this.applicationConfig = applicationConfig;
		this.mongoClient = mongoClient;
		this.objectMapper = objectMapper;
	}

	@JmsListener(destination = "${app.queue-name}")
	public void receiveMessage(String jsonMessage) {
		try {
			Message message = objectMapper.readValue(jsonMessage, Message.class);
			log.info("Received message: {} with ID: {}", message.content(), message.id());
			// Access MongoDB and perform operations
			MongoDatabase database = mongoClient.getDatabase(applicationConfig.getDatabaseName());
			MongoCollection<Document> collection = database.getCollection(applicationConfig.getCollectionName());

			// Convert MyMessage to Document
			Document doc = new Document("content", message.content()).append("id", message.id());

			// Insert Document into MongoDB
			collection.insertOne(doc);

		} catch (JsonProcessingException e) {
			log.error("Json Processing Exception {}", e.getMessage());
		}
	}
}

Lets post a message from swagger.

Lets see whether data inserted in mongodb.