Introduction

ActiveMQ is an open-source message broker developed by Apache, widely recognized as one of the most popular and powerful messaging systems. It fully supports the JMS 1.1 and J2EE 1.4 specifications, and although the JMS specification has been around for a while, it still plays a crucial role in modern J2EE applications.

Features

1. Support for cross-language clients and protocols from Java, C, C++, C#, Ruby, Perl, Python, PHP.
2. Full support for enterprise integration patterns in JMS clients and Message Broker.
3. Many advanced features like message groups, virtual destinations, wildcards, and composite destinations.
4. Full support for JMS 1.1 and J2EE 1.4, including transient, persistent, transactional, and XA messages.
5. Spring support, allowing easy embedding into Spring applications and configuration using Spring XML.
6. Designed for high-performance clustering, client-server, and peer-to-peer communication.
7. CXF and Axis support for reliable messaging in web service stacks.
8. Can be used as an in-memory JMS provider, ideal for unit testing JMS.
9. Supports pluggable transport protocols such as in-VM, TCP, SSL, NIO, UDP, multicast, JGroups, and JXTA.
10. Fast persistence using JDBC and high-performance logging.

Windows Installation

Download from the official website: https://activemq.apache.org/activemq-5014005-release

Extract the archive.

The default credentials are admin/admin.

Linux Installation

# Download the Linux version from: https://activemq.apache.org/components/classic/download/
# Extract the archive:
tar -zxvf xxxx.gz
# Navigate to the bin directory:
./activemq start   # Start ActiveMQ
# Check the process:
ps -ef | grep activemq
# Stop ActiveMQ:
./activemq stop

Docker Installation

docker search activemq
docker pull rmohr/activemq
docker run --name activemq -d -p 61616:61616 -p 8161:8161 rmohr/activemq

Access the management console at: http://localhost:8161/admin/

JMS Code Demonstration

Create a Maven project and add the following dependency:

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.9.1</version>
</dependency>

Queue Mode

Producer

import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class AppProducer {
    private static final String URL = "tcp://localhost:61616";
    private static final String QUEUE_NAME = "queue-test";

    public static void main(String[] args) throws JMSException {
        // 1. Create connection factory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);

        // 2. Create connection
        Connection connection = connectionFactory.createConnection();

        // 3. Start connection
        connection.start();

        // 4. Create session (non-transactional, auto acknowledge)
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 5. Create destination (queue)
        Destination destination = session.createQueue(QUEUE_NAME);

        // 6. Create producer
        MessageProducer producer = session.createProducer(destination);

        // 7. Send messages
        for (int i = 0; i < 100; i++) {
            TextMessage message = session.createTextMessage("test-" + i);
            producer.send(message);
            System.out.println("Sent message: " + message.getText());
        }

        // 8. Close connection
        connection.close();
    }
}

After running the producer, visit the queue management page: http://localhost:8161/admin/queues.jsp

Consumer

import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class AppConsumer {
    private static final String URL = "tcp://localhost:61616";
    private static final String QUEUE_NAME = "queue-test";

    public static void main(String[] args) throws JMSException {
        // 1. Create connection factory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);

        // 2. Create connection
        Connection connection = connectionFactory.createConnection();

        // 3. Start connection
        connection.start();

        // 4. Create session (non-transactional, auto acknowledge)
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 5. Create destination (queue)
        Destination destination = session.createQueue(QUEUE_NAME);

        // 6. Create consumer
        MessageConsumer consumer = session.createConsumer(destination);

        // 7. Set message listener
        consumer.setMessageListener(message -> {
            TextMessage textMessage = (TextMessage) message;
            try {
                System.out.println("Received message: " + textMessage.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        });

        // 8. Note: Do not close the connection; the consumer continues listening.
        // connection.close();
    }
}

Start two consumers simultaneously, then run the producer. Messages will be distributed in a load-balanecd manner.

Topic Mode

In topic mode, subscribers will not receive any messages unless they are subscribed before the messages are sent.

Producer for Topic

import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class AppProducer {
    private static final String URL = "tcp://localhost:61616";
    private static final String TOPIC_NAME = "queue-topic";

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
        Connection connection = connectionFactory.createConnection();
        connection.start();

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createTopic(TOPIC_NAME); // Topic instead of Queue
        MessageProducer producer = session.createProducer(destination);

        for (int i = 0; i < 100; i++) {
            TextMessage message = session.createTextMessage("test-" + i);
            producer.send(message);
            System.out.println("Sent message: " + message.getText());
        }

        connection.close();
    }
}

Subscriber

import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class AppConsumer2 {
    private static final String URL = "tcp://localhost:61616";
    private static final String TOPIC_NAME = "queue-topic";

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
        Connection connection = connectionFactory.createConnection();
        connection.start();

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createTopic(TOPIC_NAME); // Topic
        MessageConsumer consumer = session.createConsumer(destination);

        consumer.setMessageListener(message -> {
            TextMessage textMessage = (TextMessage) message;
            try {
                System.out.println("Received message: " + textMessage.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        });
    }
}

Start two subscribers first, then run the producer. Both subscribers will receive all messages.

Spring Boot Integration with JMS and ActiveMQ

Connection factories:

• SingleConnectionFactory: Uses only a single connection.
• CachingConnectionFactory: Extends the above and adds caching for sessions, consumers, and producers.

For a detailed implementation, refer to this blog post: https://www.cnblogs.com/elvinle/p/8457596.html