Message Broker Availability Mechanisms

Message queues ensure reliable delivery through built-in features like transactions, durable delivery, and client acknowledgments. External persistence storage provides an additional layer of fault tolerance, safeguarding data against unexpected broker crashes.

Core Persistence Logic

Persistence mechanisms prevent data loss when the broker shuts down unexpectedly. The standard storage workflow operates as follows:

• Upon receiving a message from a producer, the broker first commits it to a local disk, memory database, or remote relational database.
• The broker then attempts delivery to the consumer. If successful, the stored record is removed.
• If delivery fails, the broker continuously retries.
• During broker startup, the system checks the storage backend for any undelivered messages and prioritizes their dispatch.

Available Storage Engines

AMQ Message Store

This file-based approach offers rapid write speeds and straightforward recovery. Messages are grouped into 32MB files. Once all messages within a file are consumed, it is marked for deletion and purged during the next cleanup cycle. Note: This engine is obsolete and only applicable to ActiveMQ versions prior to 5.3.

KahaDB Storage Engine

Since ActiveMQ 5.4, KahaDB has served as the default persistence plugin. It leverages a transaction log and a single index file to track all addresses, optimizing performance and recovery capabilities for typical message consumption patterns. Data is appended to log files, and obsolete logs are discarded automatically.

The KahaDB directory structure consists of four file types and a lock file:

• db-number.log: Data records stored in sequentially numbered, fixed-size files (e.g., db-1.log, db-2.log). Unreferenced files are deleted or archived.
• db.data: Contains the persistent B-Tree index pointing to records within the log files.
• db.free: Tracks unused page IDs within the current db.data file.
• db.redo: Facilitates B-Tree index recovery if the broker forces a restart after an unclean shutdown.
• lock: Indicates exclusive write permissions for the current broker instance.

JDBC Storage

This method delegates message persistence to a third-party relational database via the JDBC protocol.

LevelDB Storage

Introduced in ActiveMQ 5.8, this file-based local database engine outperformed KahaDB in persistence speed. Instead of custom B-Tree implementations, it relied on LevelDB-based indexing. Note: The LevelDB store was officially removed in version 5.17.

JDBC Store with ActiveMQ Journal

This hybrid approach mitigates JDBC latency by utilizing a high-speed journal cache. It significantly boosts throughput by buffering writes.

To verify the current configuration, inspect the persistenceAdapter block within conf/activemq.xml.

Implementing JDBC Persistence

MySQL Integration

Deploying ActiveMQ with MySQL requires adding the corresponding driver to the broker's library directory.

# Navigate to the ActiveMQ library path
cd /opt/messaging/apache-activemq/lib
# Retrieve the MySQL connector
wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.28/mysql-connector-java-8.0.28.jar

Adapter Configuration

Modify conf/activemq.xml to replace the default KahaDB adapter with the JDBC variant:

<persistenceAdapter>
    <jdbcPersistenceAdapter dataSource="#mysqlDataSource" createTableOnStartup="true"/>
</persistenceAdapter>

The dataSource attribute links to the database connection bean. The createTableOnStartup flag dictates automatic schema generation; it should be toggled to false after the initial deployment to prevent redundant table creation.

Connection Pool Setup

Define the database pool bean within the same configuration file:

<bean id="mysqlDataSource" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">
    <property name="driverClassName" value="com.mysql.cj.jdbc.Driver"/>
    <property name="url" value="jdbc:mysql://192.168.10.50:3306/mq_store?relaxAutoCommit=true"/>
    <property name="username" value="admin"/>
    <property name="password" value="securePass123"/>
    <property name="poolPreparedStatements" value="true"/>
</bean>

Schema Initialization

Create the target database prior to launching the broker:

CREATE DATABASE mq_store;

Restart the service to trigger table generation:

./activemq stop && ./activemq start

The system automatically provisions three tables:

• ACTIVEMQ_MSGS: Holds the actual message payloads. Key columns include ID (primary key), CONTAINER (destination), MSGID_PROD (producer key), MSGID_SEQ (sequence), EXPIRATION (timeout), MSG (binary payload), and PRIORITY.
• ACTIVEMQ_ACKS: Records consumer acknowledgments to track successful deliveries and persistent subscriptions.
• ACTIVEMQ_LOCK: Manages master broker election in clustered environments.

If automatic generation fails, execute the following schema definitions manually:

CREATE TABLE ACTIVEMQ_ACKS (
    CONTAINER VARCHAR(250) NOT NULL,
    SUB_DEST VARCHAR(250) NULL,
    CLIENT_ID VARCHAR(250) NOT NULL,
    SUB_NAME VARCHAR(250) NOT NULL,
    SELECTOR VARCHAR(250) NULL,
    LAST_ACKED_ID BIGINT NULL,
    PRIORITY BIGINT DEFAULT 5 NOT NULL,
    XID VARCHAR(250) NULL,
    PRIMARY KEY (CONTAINER, CLIENT_ID, SUB_NAME, PRIORITY)
);

CREATE INDEX ACTIVEMQ_ACKS_XIDX ON ACTIVEMQ_ACKS (XID);

CREATE TABLE ACTIVEMQ_LOCK (
    ID BIGINT NOT NULL PRIMARY KEY,
    TIME BIGINT NULL,
    BROKER_NAME VARCHAR(250) NULL
);

CREATE TABLE ACTIVEMQ_MSGS (
    ID BIGINT NOT NULL PRIMARY KEY,
    CONTAINER VARCHAR(250) NOT NULL,
    MSGID_PROD VARCHAR(250) NULL,
    MSGID_SEQ BIGINT NULL,
    EXPIRATION BIGINT NULL,
    MSG BLOB NULL,
    PRIORITY BIGINT NULL,
    XID VARCHAR(250) NULL
);

CREATE INDEX ACTIVEMQ_MSGS_CIDX ON ACTIVEMQ_MSGS (CONTAINER);
CREATE INDEX ACTIVEMQ_MSGS_EIDX ON ACTIVEMQ_MSGS (EXPIRATION);
CREATE INDEX ACTIVEMQ_MSGS_MIDX ON ACTIVEMQ_MSGS (MSGID_PROD, MSGID_SEQ);
CREATE INDEX ACTIVEMQ_MSGS_PIDX ON ACTIVEMQ_MSGS (PRIORITY);
CREATE INDEX ACTIVEMQ_MSGS_XIDX ON ACTIVEMQ_MSGS (XID);

Producer Implementation

Durability must be explicitly enforced on the producer side. The consumer requires no special configuration. Below is an illustrative producer implementation for a queue destination:

package com.demo.messaging;

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

public class QueueSender {
    private static final String BROKER_URI = "nio://192.168.10.50:61616";
    private static final String DESTINATION_NAME = "mysql_queue_01";

    public static void main(String[] args) throws JMSException {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URI);
        Connection conn = factory.createConnection();
        conn.start();

        Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue targetQueue = sess.createQueue(DESTINATION_NAME);

        MessageProducer msgProducer = sess.createProducer(targetQueue);
        msgProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

        for (int idx = 1; idx <= 3; idx++) {
            TextMessage payload = sess.createTextMessage("Database persistent message #" + idx);
            msgProducer.send(payload);
        }

        msgProducer.close();
        sess.close();
        conn.close();
    }
}

Queue vs Topic Behavior

• Queue: Unconsumed messages remain in ACTIVEMQ_MSGS. Once any consumer processes a message, the corresponding row is deleted.
• Topic: For durable subscribers, relationships are preserved permanently in ACTIVEMQ_ACKS, and messages persist in ACTIVEMQ_MSGS. The LAST_ACKED_ID column maps to the message ID, identifying the last processed record for each subscriber.

Common Configuration Pitfalls

• Missing JDBC driver or connection pool jars in the lib folder.
• Failing to disable createTableOnStartup post-initialization, leading to schema overwrite risks.
• Throwing IllegalStateException: LifecycleProcessor not initialized due to underscores in the machine's hostname; renaming the host resolves this.

Implementing JDBC with Journal Cache

The pure JDBC adapter incurs database read/write overhead per message. The Journal-enhanced variant utilizes a transient write cache to drastically amplify performance. When consumers process messages rapidly, the journal bypasses database synchronization entirely for consumed payloads. For slower consumption rates, the journal batches writes to the database efficiently. Data is not immediately flushed to the relational store.

Journal Adapter Configuration

Update conf/activemq.xml to employ the journal persistence factory:

<persistenceFactory>
    <journalPersistenceAdapterFactory
        journalLogFiles="5"
        journalLogFileSize="32768"
        useJournal="true"
        useQuickJournal="true"
        dataSource="#mysqlDataSource"
        dataDirectory="../activemq-data" />
</persistenceFactory>

Pair this with the identical MySQL connection pool bean used for standalone JDBC persistence, and restart the broker.

Historical Evolution of Persistence Engines

ActiveMQ persistence strategies have evolved significantly over time. The initial AMQ file store was supplemented by High Performance Journal attachments and relational database support in version 4. KahaDB emerged in version 5.3, supplanting AMQ as the default by 5.4. LevelDB was integrated starting from 5.8, leading to the standard ZooKeeper + Replicated LevelDB master-slave clustering architecture introduced in version 5.9.