JMS (Java Message Service) Fundamentals

JMS is a Java API specification that allows applications to interact with Message-Oriented Middleware (MOM) systems. MOM provides a reliable messaging mechanism for data exchange between distributed systems, enabling platform-independent communication. Popular MOM implementations include Apache ActiveMQ, Alibaba's RocketMQ, IBM MQSeries, Microsoft MSMQ, and BEA RabbitMQ. Not all MOM systems comply with the JMS specification. A MOM system built on JMS is called a JMS Provider.

A "message" is a unit of data transmitted between computers. It can be as simple as a text string or more complex, containing embedded objects. Messages are sent to queues.

A "message queue" is a container that holds messages during transmission. The queue manager acts as an intermediary, routing messages from source to destination. Queues provide routing and guarantee message delivery; if the receiver is unavailable when a message is sent, the queue retains the message until successful delivery is possible.

Key characteristics of message queues include asynchronous processing, which reduces request response time and decouples components. Typical use cases involve operations that are time-consuming and do not require immediate (synchronous) results. Since message formats are fixed, senders and receivers do not need direct contact or mutual influence, achieving decoupling. Examples include asynchronous communication across systems (like sending text messages or emails instead of phone calls), loose coupling between multiple applications (platform- and language-independent, not function calls), enterprise application integration (EAI) via message queues instead of file transfer or RPC, converting synchronous operations within applications to asynchronous (e.g., order processing where front-end apps place orders into a queue, and back-end apps process them sequentially), event-driven architectures (EDA) where processing flows are split into stages connected by queues, flexible coupling like publish/subscribe with routing rules, and cross-LAN or cross-city communication.

Message Broker Application Scenarios

Asynchronous Communication

Some operations do not need immediate handling. Message queues provide an asynchronous mechanism: a message is placed into a queue without immediate processing, allowing accumulation and later handling.

Buffering

Different system elements may require different processing times. A message queue acts as a buffer layer to optimize data flow speed and control system response time.

Decoupling

Message queues reduce strong dependencies between engineering components and accommodate heterogeneous systems. An implicit, data-based interface layer is inserted between processing stages, allowing independant extension or modification as long as both sides adhere to the same interface.

Redundancy

Data processing can fail. Without persistence, data may be lost. Message queues persist data until fully processed, using patterns like "insert-get-delete". The processing system must explicitly report message completion before the queue deletes it, ensuring data safety.

Scalability

Because message queues decouple processing, increasing message enqueue and processing frequency is straightforward by adding more processing processes. No code changes or parameter adjustments are needed, facilitating distributed scaling.

Recoverability

Failure of one system component does not affect the whole system. Message queues reduce inter-process coupling, so if a message processing process fails, queued messages can still be processed after recovery.

Ordering Guarantee

In many scenarios, data processing order matters. Most message queues are inherently ordered and guarantee data processing in a specific sequence.

Overload Protection

Sudden traffic spikes can be unpredictable. Instead of provisioning resources for peak loads, message queues allow key components to withstand bursts without crashing from overload.

Data Stream Processing

Distributed systems generate massive data streams like business logs, monitoring data, and user behavior. Real-time or batch collection and analysis of these streams is essential in modern internet systems, and message queues are ideal for such data aggregation.

Comparison of Common Message Queues (ActiveMQ, RabbitMQ, RocketMQ, Kafka)

Feature	ActiveMQ	RabbitMQ	RocketMQ	Kafka
Producer/Consumer	Yes	Yes	Yes	Yes
Publish/Subscribe	Yes	Yes	Yes	Yes
Request/Reply	Yes	Yes	No	No
API Completeness	High	High	High	High
Multi-language Support	Yes	Yes	Java	Yes
Single-node Throughput	Tens of thousands	Tens of thousands	Tens of thousands	Hundreds of thousands
Message Latency	None	Microseconds	Milliseconds	Milliseconds
Availability	High (Master/Slave)	High (Master/Slave)	Very High (Distributed)	Very High (Distributed)
Message Loss	Low	Low	Theoretically none	Theoretically none
Documentation Completeness	High	High	High	High
Quick Start Provided	Yes	Yes	Yes	Yes
Community Activity	High	High	Yes	High
Commercial Support	None	None	Commercial Cloud	Commercial Cloud

JMS Roles

Broker

The message server that provides core messaging services.

Provider (Producer)

Created by a session, sends messages to a destination.

Consumer

Created by a session, receives messages from a destination. Two consumption methods exist:

• Synchronous: The consumer calls receive() to extract messages, blocking until a message arrives.
• Asynchronous: A message listener is registered, defining actions upon message arrival.

P2P (Point-to-Point) Model

A producer sends messages to a queue. A consumer takes and processes messages. After consumption, the message is removed from the queue. Multiple consumers can exist, but only one can consume a specific message. If no consumer is active, messages are retained until one appears.

Pub/Sub (Publish/Subscribe) Model

A producer (publisher) sends messages to a topic. Multiple consumers (subscribers) receive the message. Unlike P2P, all subscribers get the message. Messages are not persisted unless a durable subscription is used; subscribers must be active before the message is published to receive it.

P2P vs Pub/Sub Quick Comparison

Aspect	Topic	Queue
Messaging pattern	Publish/Subscribe	Point-to-Point
State	Default: no persistence, stateless	Default: persisted as files (e.g., KahaDB)
Delivery guarantee	Not all subscribers guaranteed to receive	Each message guaranteed to be received (if no timeout)
Message loss	If no subscriber active, message lost	Messages not lost; retained until consumed (unless timed out)
Distribution strategy	One-to-many; all subscribers receive	One-to-one; only one consumer receives

Queue

Used primarily in P2P mode. Typically consumed by a single consumer; message removed after processing.

Topic

Used in Pub/Sub mode. Messages broadcast to all subscribers. Suitable for scenarios where messages can be processed multiple times.

Both Queue and Topic implement the Destination interface.

ConnectionFactory

Creates connections. Example: ActiveMQConnectionFactory.

Connection

Encapsulates a virtual connection between client and JMS provider.

Destination

Specifies the target for message production or source for consumption. Two domains exist:

• P2P: Destination is a queue; each message has one consumer; no temporal dependency between producer and consumer.
• Pub/Sub: Destination is a topic; each message can have multiple consumers; temporal dependency exists (consumer must subscribe before message publication). Durable subscripsions relax this requirement.

Session

A single-threaded context for producing and consuming messages. Creates producers, consumers, and messages. Provides a transactional context for atomic operations.

JMS Message Structure

A JMS message consists of three parts:

• Header: Each header field has getter/setter methods.
• Properties: Additional custom values beyond headers.
• Body: Message content types include:
  • TextMessage: Text data.
  • MapMessage: Key/value pairs.
  • BytesMessage: Byte stream.
  • StreamMessage: Java primitive data stream.
  • ObjectMessage: Serializable Java object.

Message Reliability Mechanisms

Acknowledgment

A message is considered successfully consumed only after acknowledgment. The process includes: client receives, client processes, and message is acknowledged.

• In a transactional session: acknowledgment occurs when the transaction commits.
• In a non-transactional session: depends on the acknowledgment mode:
  • Session.AUTO_ACKNOWLEDGE: Message automatically acknowledged when receive() returns or onMessage() completes successfully.
  • Session.CLIENT_ACKNOWLEDGE: Client calls acknowledge() on the message. Acknowledging one message acknowledges all previously consumed messages in the session.
  • Session.DUPS_OK_ACKNOWLEDGE: Session lazily acknowledges messages, potentially allowing duplicates; the broker sets JMSRedelivered to true for duplicates.

Persistence

Two delivery modes:

• PERSISTENT: Broker persists the message to survive broker failure.
• NON_PERSISTENT: No persistence required.

Priority

Ten levels (0-9, default 4). Higher priority messages are delivered first, but not guaranteed strictly.

Message Expiration

Messages can have a time-to-live; default is no expiration.

Temporary Destinations

Created via createTemporaryQueue or createTemporaryTopic on a session. They live only as long as the connection. Only the creating connection's consumers can access them.

Durable Subscription

The producer must use PERSISTENT mode. The client creates a durable subscription via createDurableSubscriber on a topic. The broker stores messages for the subscription even when the subscriber is inactive. When the subscriber reconnects with the same client ID, connection factory, topic, and subscription name, it receives stored messages. Only one active subscriber per durable subscription at a time. Durable subscriptions persist until unsubscribe() is called.

Local Transactions

JMS clients can group message send/receive operations into local transactions using commit() and rollback() on a session. Commit sends all produced messages and acknowledges all consumed messages; rollback destroys produced messages and recovers consumed ones. Transactional sessions are always in a transaction; closing rolls back the current transaction. Note: Do not send a message and wait for its reply in the same transaction (deadlock). Also, message production and consumption cannot be in the same transaction.

ActiveMQ

Official website: http://activemq.apache.org/

Broker

In the ActiveMQ 5.0 binary distribution, the bin directory contains the activemq script. Running it starts a broker. Brokers can be configured via Broker Configuration URI or Broker XBean URI. Examples:

Example	Description
activemq	Uses default xbean:activemq.xml config file
activemq xbean:myconfig.xml	Uses myconfig.xml from classpath
activemq xbean:file:./conf/broker1.xml	Uses broker1.xml relative path
activemq xbean:file:C:/ActiveMQ/conf/broker2.xml	Uses broker2.xml absolute path
activemq broker:(tcp://localhost:61616, tcp://localhost:5000)?useJmx=true	Broker with two transport connectors and JMX enabled
activemq broker:(tcp://localhost:61616, network:tcp://localhost:5000)?persistent=false	Broker with transport and network connector, persistence disabled

Storage

KahaDB (Default)

Messages are appended sequentially to a log file. An index file (B-Tree) points to message locations. A transaction log supports recovery. Optimized for typical messaging patterns.

In data/kahadb directory, four files are created:

1. db.data: Message index file (B-Tree).
2. db.redo: For message recovery.
3. db-*.log: Message content; new data appended sequentially (fast writes). Default max file size 32MB.
4. lock: Lock file for cluster contention.

Configuration example (activemq.xml):

<persistenceAdapter>
    <kahaDB directory="${activemq.data}/kahadb" journalMaxFileLength="16mb"/>
</persistenceAdapter>

Features:

• Log-style storage.
• B-Tree index for fast updates.
• Full JMS transaction support.
• Multiple recovery mechanisms.

AMQ Storage (pre-5.3)

File-based database; messages stored in files with memory caching. Higher performance than JDBC but heavy index file usage and slow index rebuild on crash. Not recommended.

JDBC Storage

Persists to a database. Three tables created by default:

• activemq_msgs: Stores queue and topic messages.
• activemq_acks: Stores durable subscription info and last consumed message ID.
• activemq_lock: Ensures only one broker accesses the database at a time (cluster).

Example configuration (MySQL data source):

<bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
    <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
    <property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>
    <property name="username" value="activemq"/>
    <property name="password" value="activemq"/>
    <property name="maxActive" value="200"/>
    <property name="poolPreparedStatements" value="true"/>
</bean>

<persistenceAdapter>
    <jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="true" />
</persistenceAdapter>

Required JARs: commons-dbcp, commons-pool, mysql-connector-java.

Table Fields

activemq_acks: Stores subscriptions for durable topics.

• container: Destination.
• sub_dest: Cluster info for static clusters.
• client_id: Unique client ID per subscriber.
• sub_name: Subscriber name.
• selector: Filter conditions.
• last_acked_id: Last consumed message ID.

activemq_lock: Used in clusters to designate the master broker.

activemq_msgs: Stores messages.

• id: Auto-increment primary key.
• container: Destination.
• msgid_prod: Producer's client ID.
• msg_seq: Message sequence; combined with msgid_prod forms JMSMessageID.
• expiration: Expiry timestamp (epoch milliseconds).
• msg: Serialized message body (binary).
• priority: 0-9.
• xid: Used for subscription info.

LevelDB Storage

Higher performance than KahaDB, but officially deprecated; KahaDB remains default and recommended.

Memory Storage

When persistent="false", messages stored in memory. Set on the broker element.

JDBC with ActiveMQ Journal

Delays writing to DB by caching messages in a journal file first. If consumer speed is high, many messages are consumed before journal sync, reducing DB writes. If consumer is slow, batch writes occur.

Protocols

ActiveMQ supports multiple client-broker transport protocols: TCP, NIO, UDP, SSL, HTTP(S), VM.

Transmission Control Protocol (TCP)

• Default; port 61616.
• Data serialized via OpenWire wire protocol.
• URI: tcp://hostname:port?key=value.
• Advantages: reliable, efficient, widely supported.

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>

New I/O API Protocol (NIO)

• Similar to TCP but uses fewer threads, better for many concurrent clients or slow networks.
• URI: nio://hostname:port?key=value.

<transportConnectors>
    <transportConnector name="tcp" uri="tcp://localhost:61616?trace=true" />
    <transportConnector name="nio" uri="nio://localhost:61618?trace=true" />
</transportConnectors>

User Datagram Protocol (UDP)

• Unreliable, no delivery guarantee. Suitable for fast data where loss is acceptable, or through firewalls.
• URI: udp://hostname:port?key=value.

<transportConnector name="udp" uri="udp://localhost:61618?trace=true" />

Secure Sockets Layer Protocol (SSL)

• Encrypted transport.
• URI: ssl://hostname:port?key=value.

<transportConnector name="ssl" uri="ssl://localhost:61617?trace=true" />

Hypertext Transfer Protocol (HTTP/HTTPS)

• For firewall-friendly access.
• URI: http://hostname:port?key=value or https://....

<transportConnector name="http" uri="http://localhost:8080?trace=true" />

VM Protocol

• In-VM communication, no network overhead; uses direct method calls.
• First connection creates an embedded broker; all subsequent connections with same broker name use it. Broker auto-stops when all connections close.
• URI: vm://brokerName?key=value.
• Embedded example: vm:broker:(tcp://localhost:6000)?brokerName=embeddedbroker&persistent=false.
• Config file: vm://localhost?brokerConfig=xbean:activemq.xml.

HelloWorld Example

Download

From http://activemq.apache.org/.

Installation & Startup

Extract and run bin/win64/activemq.bat (Windows).

Web Console

Access http://localhost:8161/.

Change Web Console Port

Edit conf/jetty.xml in ActiveMQ installation directory, modify the jettyport element, then restart.

Maven Dependency

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

Sender Code

package com.example.activemq;

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

public class Sender {
    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
                ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                "tcp://localhost:61616");
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("example.queue");
        MessageProducer producer = session.createProducer(queue);

        for (int i = 0; i < 100; i++) {
            TextMessage msg = session.createTextMessage("Hello " + i);
            Thread.sleep(1000);
            producer.send(msg);
        }

        connection.close();
        System.out.println("Messages sent.");
    }
}

Receiver Code

package com.example.activemq;

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

public class Receiver {
    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
                ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                "tcp://localhost:61616");
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("example.queue");
        MessageConsumer consumer = session.createConsumer(queue);

        while (true) {
            TextMessage received = (TextMessage) consumer.receive();
            System.out.println("Received: " + received.getText());
        }
    }
}

ActiveMQ Security Mechanisms

Web Console Security

Edit conf/jetty-realm.properties:

admin: admin, admin
user: user, user
yiming: 123, user

Format: username: password, role. Restart ActiveMQ for changes.

Message Security

Add the following inside the <broker> element in activemq.xml:

<plugins>
    <simpleAuthenticationPlugin>
        <users>
            <authenticationUser username="admin" password="admin" groups="admins,publishers,consumers"/>
            <authenticationUser username="publisher" password="publisher" groups="publishers,consumers"/>
            <authenticationUser username="consumer" password="consumer" groups="consumers"/>
            <authenticationUser username="guest" password="guest" groups="guests"/>
        </users>
    </simpleAuthenticationPlugin>
</plugins>