Apache ActiveMQ Core Knowledge

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: activemq for comprehensive documentation.

Quick Start (Docker)

docker-compose.yml

services: activemq: image: apache/activemq-artemis:latest ports: - "61616:61616" # AMQP/OpenWire - "8161:8161" # Web Console environment: - ARTEMIS_USER=admin - ARTEMIS_PASSWORD=admin volumes: - activemq_data:/var/lib/artemis-instance

volumes: activemq_data:

docker-compose up -d

Web Console: http://localhost:8161/console

Core Concepts

Concept Description

Queue Point-to-point messaging

Topic Publish-subscribe messaging

Selector SQL-like message filtering

Durable Subscriber Persisted topic subscription

Message Groups Ordered message processing

Virtual Topic Queue semantics on topics

JMS Architecture

┌─────────────────────────────────────────────────────────┐ │ ActiveMQ Broker │ │ │ │ Queue (P2P) Topic (Pub/Sub) │ │ ┌─────────┐ ┌─────────┐ │ │ │ Message │ │ Message │ │ │ │ Queue │──▶Consumer │ Topic │──▶Subscriber 1 │ │ └─────────┘ └─────────┘──▶Subscriber 2 │ │ ▲ ▲ │ │ │ │ │ │ Producer Publisher │ └─────────────────────────────────────────────────────────┘

JMS Message Types

Type Description

TextMessage

String content

BytesMessage

Binary data

MapMessage

Key-value pairs

ObjectMessage

Serialized Java object

StreamMessage

Sequential data stream

Producer Patterns

Java (Spring JMS)

@Configuration @EnableJms public class JmsConfig { @Bean public ConnectionFactory connectionFactory() { ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(); factory.setBrokerURL("tcp://localhost:61616"); factory.setUserName("admin"); factory.setPassword("admin"); return factory; }

@Bean
public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
    JmsTemplate template = new JmsTemplate(connectionFactory);
    template.setDeliveryPersistent(true);
    template.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
    return template;
}

@Bean
public JmsTemplate topicJmsTemplate(ConnectionFactory connectionFactory) {
    JmsTemplate template = new JmsTemplate(connectionFactory);
    template.setPubSubDomain(true);  // Enable topic mode
    return template;
}

}

@Service public class OrderProducer { @Autowired private JmsTemplate jmsTemplate;

// Send to queue
public void sendToQueue(Order order) {
    jmsTemplate.convertAndSend("orders.queue", order, message -> {
        message.setJMSCorrelationID(UUID.randomUUID().toString());
        message.setStringProperty("orderType", order.getType());
        message.setIntProperty("priority", order.getPriority());
        return message;
    });
}

// Send with reply
public OrderResponse sendAndReceive(Order order) {
    return (OrderResponse) jmsTemplate.sendAndReceive("orders.queue",
        session -> {
            ObjectMessage msg = session.createObjectMessage(order);
            msg.setJMSReplyTo(session.createTemporaryQueue());
            return msg;
        });
}

}

@Service public class EventPublisher { @Autowired @Qualifier("topicJmsTemplate") private JmsTemplate topicTemplate;

// Publish to topic
public void publishEvent(OrderEvent event) {
    topicTemplate.convertAndSend("orders.events", event);
}

}

Node.js (stompit)

import stompit from 'stompit';

const connectOptions = { host: 'localhost', port: 61613, connectHeaders: { host: '/', login: 'admin', passcode: 'admin', 'heart-beat': '5000,5000', }, };

stompit.connect(connectOptions, (error, client) => { if (error) { console.error('Connection error:', error); return; }

const sendHeaders = { destination: '/queue/orders', 'content-type': 'application/json', persistent: 'true', 'correlation-id': uuidv4(), };

const frame = client.send(sendHeaders); frame.write(JSON.stringify(order)); frame.end();

client.disconnect(); });

Python (stomp.py)

import stomp import json

class OrderListener(stomp.ConnectionListener): def on_error(self, frame): print(f'Error: {frame.body}')

def on_message(self, frame):
    print(f'Received: {frame.body}')

conn = stomp.Connection([('localhost', 61613)]) conn.set_listener('', OrderListener()) conn.connect('admin', 'admin', wait=True)

Send to queue

conn.send( destination='/queue/orders', body=json.dumps(order), headers={ 'persistent': 'true', 'content-type': 'application/json', 'correlation-id': correlation_id, } )

conn.disconnect()

Consumer Patterns

Java (Spring JMS)

@Configuration public class JmsListenerConfig { @Bean public DefaultJmsListenerContainerFactory jmsListenerContainerFactory( ConnectionFactory connectionFactory) { DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setConcurrency("3-10"); factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE); factory.setErrorHandler(t -> log.error("JMS Error", t)); return factory; }

@Bean
public DefaultJmsListenerContainerFactory topicListenerContainerFactory(
        ConnectionFactory connectionFactory) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setPubSubDomain(true);
    factory.setSubscriptionDurable(true);
    factory.setClientId("order-service");
    return factory;
}

}

@Service public class OrderConsumer { // Queue consumer @JmsListener(destination = "orders.queue", concurrency = "3-10") public void consumeQueue( @Payload Order order, @Header(JmsHeaders.CORRELATION_ID) String correlationId, @Header(name = "orderType", required = false) String orderType, Session session, Message message) throws JMSException {

    try {
        processOrder(order);
        message.acknowledge();
    } catch (Exception e) {
        session.recover();  // Redelivery
    }
}

// Queue consumer with selector
@JmsListener(
    destination = "orders.queue",
    selector = "orderType = 'EXPRESS' AND priority > 5"
)
public void consumeExpressOrders(Order order) {
    processExpressOrder(order);
}

// Durable topic subscriber
@JmsListener(
    destination = "orders.events",
    containerFactory = "topicListenerContainerFactory",
    subscription = "order-processor"
)
public void consumeTopic(OrderEvent event) {
    processEvent(event);
}

}

Request/Reply Pattern

@Service public class OrderService { @JmsListener(destination = "orders.request") @SendTo("orders.response") public OrderResponse processOrderRequest(Order order) { // Process and return response return new OrderResponse(order.getId(), "PROCESSED"); } }

Node.js (stompit)

stompit.connect(connectOptions, (error, client) => { const subscribeHeaders = { destination: '/queue/orders', ack: 'client-individual', };

client.subscribe(subscribeHeaders, (error, message) => { if (error) { console.error('Subscribe error:', error); return; }

message.readString('utf-8', (error, body) => {
  if (error) {
    client.nack(message);
    return;
  }

  try {
    const order = JSON.parse(body);
    processOrder(order);
    client.ack(message);
  } catch (e) {
    client.nack(message);
  }
});

}); });

Message Groups

Ensures ordered processing by grouping related messages.

// Producer jmsTemplate.convertAndSend("orders.queue", order, message -> { message.setStringProperty("JMSXGroupID", order.getCustomerId()); message.setIntProperty("JMSXGroupSeq", sequenceNumber); return message; });

// Consumer receives messages in order per group @JmsListener(destination = "orders.queue") public void consume(Order order, @Header("JMSXGroupID") String groupId) { // Messages with same groupId processed in order }

Virtual Topics

Combine topic semantics with queue load balancing.

Publisher ──▶ VirtualTopic.Orders ──▶ Consumer.A.VirtualTopic.Orders (Consumer Group A) ──▶ Consumer.B.VirtualTopic.Orders (Consumer Group B)

// Publisher sends to virtual topic topicTemplate.convertAndSend("VirtualTopic.Orders", event);

// Consumer subscribes to virtual consumer queue @JmsListener(destination = "Consumer.OrderProcessor.VirtualTopic.Orders") public void consume(OrderEvent event) { // Load balanced within consumer group }

When NOT to Use This Skill

Use alternative messaging solutions when:

• Event streaming with replay - Kafka provides better event sourcing

• Ultra-high throughput - Kafka or Pulsar scale better

• Cloud-native microservices - NATS or cloud providers are simpler

• Non-Java environments primarily - RabbitMQ has better multi-language support

• Serverless architecture - Use cloud-native queues (SQS, Service Bus)

• Simple pub/sub - Redis or NATS may be sufficient

• Modern streaming patterns - Kafka Streams or Pulsar Functions

Anti-Patterns

Anti-Pattern Why It's Bad Solution

Non-persistent delivery in production Message loss on broker restart Use persistent delivery mode

No DLQ configured Poison messages loop forever Configure dead letter queue

Auto-acknowledge mode Message loss on consumer crash Use client or transacted acknowledge

Large message payloads Memory pressure, poor performance Store in database/blob, send reference

Too many topics/queues Management overhead Use selectors or consolidate destinations

ObjectMessage serialization Security risks, coupling Use JSON/XML with TextMessage

No prefetch limit Consumer overwhelmed Configure prefetch based on processing time

Synchronous sends everywhere Poor throughput Use async sends for non-critical messages

No connection pooling Too many connections Use pooled connection factory

Default persistence adapter Poor performance at scale Use KahaDB or JDBC with tuning

Quick Troubleshooting

Issue Likely Cause Fix

Messages piling up Slow/dead consumers Check consumer health and count

High memory usage Too many messages in memory Enable message cursor, reduce prefetch

Connection refused Wrong URL or broker down Verify broker URL and broker status

Messages not consumed No active consumers Start consumers, check subscriptions

Duplicate messages Non-transacted consumers Use transactions or implement idempotency

Store full Message retention too long Configure message TTL, purge old messages

Slow message delivery Persistence bottleneck Tune KahaDB, use faster disk

Selector not working Wrong SQL syntax Test selector syntax, check message properties

Client timeout Slow network or broker overload Increase socket timeout, scale broker

Durable subscriber not receiving Subscription disconnected Check client ID and subscription name

Production Readiness

Security Configuration

<!-- broker.xml (Artemis) --> <security-settings> <security-setting match="#"> <permission type="createNonDurableQueue" roles="admin,user"/> <permission type="deleteNonDurableQueue" roles="admin,user"/> <permission type="createDurableQueue" roles="admin"/> <permission type="deleteDurableQueue" roles="admin"/> <permission type="consume" roles="admin,user"/> <permission type="send" roles="admin,user"/> </security-setting> </security-settings>

<acceptors> <acceptor name="ssl"> tcp://0.0.0.0:61617?sslEnabled=true;keyStorePath=/path/keystore.jks;keyStorePassword=password </acceptor> </acceptors>

// Client with SSL ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(); factory.setBrokerURL("ssl://localhost:61617"); factory.setTrustStore("/path/truststore.jks"); factory.setTrustStorePassword("password");

High Availability

<!-- Live-backup pair --> <ha-policy> <replication> <master> <group-name>live-backup-group</group-name> <check-for-live-server>true</check-for-live-server> </master> </replication> </ha-policy>

Monitoring Metrics

Metric Alert Threshold

Queue depth

10000 messages

Consumer count < expected

Memory usage

80%

Store usage

80%

Dispatch rate Anomaly detection

Enqueue rate Anomaly detection

Checklist

• SSL/TLS enabled

• Authentication configured

• Authorization (JAAS) set up

• Dead letter queue configured

• Message expiration set

• Persistence enabled

• Clustering configured

• JMX monitoring enabled

• Disk usage alerts

• Backup procedures tested

Reference Documentation

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: activemq for comprehensive documentation.

Available topics: basics , destinations , producers , consumers , spring-integration , production