Table of contents · Introduction to ActiveMQ · Installing ActiveMQ · Message-oriented middleware · JMS specification · What's the Java Message Service ...
Table of contents
· What's the Java Message Service?
· How are messages stored by ActiveMQ?
· Caching messages in the broker for consumers
· Embedding ActiveMQ using Spring
Introduction to ActiveMQ
1. ActiveMQ is an open source, Java Message Service (JMS) 1.1–compliant, message-oriented middleware (MOM) from the Apache Software Foundation that provides high availability, performance, scalability, reliability, and security for enterprise messaging.
2. ActiveMQ features:
a) JMS compliance.
b) Connectivity - ActiveMQ supports for protocols such as HTTP/S, IPmulticast, SSL, STOMP, TCP, UDP, XMPP, and more.
c) Pluggable persistence and security - ActiveMQ provides multiple flavors of persistence. Security can be completely customized for the type of authentication and authorization.
d) Integration with application servers - It’s common to integrate ActiveMQ with a Java application server, including Apache Tomcat, Jetty, Apache Geronimo, JBoss and etc.
e) Client APIs - ActiveMQ provides client APIs for many languages besides just Java, including C/C++, .NET, Perl, PHP, Python, Ruby, and more.
f) Broker clustering - Many ActiveMQ brokers can work together as a federated network of brokers for scalability purposes.
g) Dramatically simplified administration - ActiveMQ is designed with developers in mind.
3. When to use ActiveMQ.
a) Heterogeneous application integration - When integrating applications written in different languages on different platforms, the various client APIs make it possible to send and receive messages via ActiveMQ no matter what language is used.
b) As a replacement for RPC - Systems that rely upon synchronous requests typically have a limited ability to scale because eventually requests will begin to back up, thereby slowing the whole system. Instead of experiencing this type of a slowdown, using asynchronous messaging, additional message receivers can be easily added so that messages are consumed concurrently and therefore handled faster.
c) To loosen the coupling between applications - A loosely coupled design is considered to be asynchronous, where the calls from either application have no bearing on one another; there’s no interdependence or timing requirements. It’s often said that applications sending messages just fire-and-forget - they send the message to ActiveMQand aren’t concerned with how or when the message is delivered. In the same manner, the consuming applications have no concern with where the messages originated or how they were sent to ActiveMQ. With a tightly coupled system design, moving a application to a new location is difficult because all segments of the application must experience an outage. With an application designed using loose coupling, different segments of the system can be moved independent of one another.
d) As the backbone of an event-driven architecture - Asynchronous processes are what afford massive scalability and high availability. When a user makes a purchase on Amazon, there are quite a few separate stages through which that order must travel including order placement, invoice creation, payment processing, order fulfillment, shipping, and more. When the user places the order, there’s a synchronous call to submit the order, but the entire order process doesn’t take place behind a synchronous call via the web browser. Instead, the order is accepted and acknowledged immediately. The rest of the steps in the process are handled asynchronously. If a problem occurs that prevents the process from proceeding, the user is notified via email.
e) To improve application scalability - The ability to design an application using a service for a specific task is the backbone of service-oriented architecture (SOA). Each service fulfills a discrete function and only that function. Then applications are built through the composition of these services, and the communication among services is achieved using asynchronous messaging and eventual consistency.
Installing ActiveMQ
1. Install JDK (version >= 1.7) and set JAVA_HOME and PATH environment variable.
2. Extract the files from the zipped tarball into a directory of your choice.
tar zxvf apache-activemq-5.13.3-bin.tar.gz cd apache-activemq-5.13.3
3. Run ActiveMQ as a daemon process.
bin/activemq start
4. Testing the Installation
a) Open the administrative interface.
i. URL: http://127.0.0.1:8161/admin/
ii. Login: admin
iii. Passwort: admin
b) Navigate to "Queues".
c) Add a queue name and click "Create".
d) Send test message by klicking on "Send to".
5. Shutdown ActiveMQ.
bin/activemq stop
Message-oriented middleware
1. Message-oriented middleware (MOM) is best described as a category of software for communication in an asynchronous, loosely-coupled, reliable, scalable, and secure manner among distributed applications or systems.
2. MOMs are important concept in the distributed computing world. They allow application-to-application communication using APIs provided by each vendor, and begin to deal with many issues in the distributed system space. A MOM acts as a message mediator between message senders and message receivers.
JMS specification
What's the Java Message Service?
1. The Java Message Service (JMS) aims to provide a standardized API to send and receive messages using the Java programming language in a vendor-neutral manner.
2. JMSisn’t itself a MOM. It’s an APIthat abstracts the interaction between messaging clients and MOMs in the same manner that JDBCabstracts communication with relational databases.
JMS client
JMS clients utilize the JMS APIfor interacting with the JMSprovider.
Non-JMS client
A non-JMS client uses a JMS provider’s native client APIinstead of the JMS API.
JMS producer
1. JMS clients use the JMS MessageProducer class for sending messages to a destination.
2. The default destination for a given producer is set when the producer is created using the Session.createProducer() method. But this can be overridden for individual messages by using the MessageProducer.send() method.
1 public interface MessageProducer { 2 void setDisableMessageID(boolean value) throws JMSException; 3 boolean getDisableMessageID() throws JMSException; 4 void setDisableMessageTimestamp(boolean value) throws JMSException; 5 boolean getDisableMessageTimestamp() throws JMSException; 6 void setDeliveryMode(int deliveryMode) throws JMSException; 7 int getDeliveryMode() throws JMSException; 8 void setPriority(int defaultPriority) throws JMSException; 9 int getPriority() throws JMSException; 10 void setTimeToLive(long timeToLive) throws JMSException; 11 long getTimeToLive() throws JMSException; 12 Destination getDestination() throws JMSException; 13 void close() throws JMSException; 14 void send(Message message) throws JMSException; 15 void send(Message message, int deliveryMode, int priority, 16 long timeToLive) 17 throws JMSException; 18 void send(Destination destination, Message message) 19 throws JMSException; 20 void send( 21 Destination destination, 22 Message message, 23 int deliveryMode, 24 int priority, 25 long timeToLive) throws JMSException; 26 }
JSM consumer
1. JMS clients use the JMS MessageConsumer class for consuming messages from a destination.
2. The MessageConsumer can consume messages either synchronously by using one of the receive() methods or asynchronously by providing a MessageListener implementation to the consumer.
1 public interface MessageConsumer { 2 String getMessageSelector() throws JMSException; 3 MessageListener getMessageListener() throws JMSException; 4 void setMessageListener(MessageListener listener) throws JMSException; 5 Message receive() throws JMSException; 6 Message receive(long timeout) throws JMSException; 7 Message receiveNoWait() throws JMSException; 8 void close() throws JMSException; 9 }
JSM provider
The JMS provider is the vendor-specific MOM that implements the JMS API.
JMS message
1. A JMS message allows anything to be sent as part of the message, including text and binary data as well as information in the headers.
2. Headers set automatically by the client’s send() method:
a) JMSDestination.
b) JMSDeliveryMode - JMS supports two types of delivery modes for messages: persistent and nonpersistent. The default delivery mode is persistent.
i. Persistent - Advises the JMS provider to persist the message so it’s not lost if the provider fails. A JMS provider must deliver a persistent message once and only once.
ii. Nonpersistent - Instructs the JMS provider not to persist the message. A JMS provider must deliver a nonpersistent message at most once. Nonpersistent messages are typically usedfor sending notifications or real-time data.
c) JMSExpiration - The time that a message will expire. The JMSExpiration message header is calculated by adding the time-to-live to the current time in GMT. By default the time-to-live is zero, meaning that the message won’t expire.
d) JMSMessageID - A string that uniquely identifies a message that’s assigned by the JMS provider and must begin with ID.
e) JMSPriority.
f) JMSTimestamp - This header denotes the time the message was sent by the producer to the JMS provider.
3. Header set optionally by the client:
a) JMSCorrelationID - Used to associate the current message with a previous message.
b) JMSReplyTo - Used to specify a destination where a reply should be sent.
c) JMSType - Used to semantically identify the message type.
4. Headers set optionally by the JMS provider:
a) JMSRedelivered - Used to indicate the liklihood that a message was previously delivered but not acknowledged.
1 public interface Message { 2 ... 3 boolean getBooleanProperty(String name) throws JMSException; 4 byte getByteProperty(String name) throws JMSException; 5 short getShortProperty(String name) throws JMSException; 6 int getIntProperty(String name) throws JMSException; 7 long getLongProperty(String name) throws JMSException; 8 float getFloatProperty(String name) throws JMSException; 9 double getDoubleProperty(String name) throws JMSException; 10 String getStringProperty(String name) throws JMSException; 11 Object getObjectProperty(String name) throws JMSException; 12 ... 13 Enumeration getPropertyNames() throws JMSException; 14 boolean propertyExists(String name) throws JMSException; 15 ... 16 void setBooleanProperty(String name, boolean value) throws JMSException; 17 void setByteProperty(String name, byte value) throws JMSException; 18 void setShortProperty(String name, short value) throws JMSException; 19 void setIntProperty(String name, int value) throws JMSException; 20 void setLongProperty(String name, long value) throws JMSException; 21 void setFloatProperty(String name, float value) throws JMSException; 22 void setDoubleProperty(String name, double value) throws JMSException; 23 void setStringProperty(String name, String value) throws JMSException; 24 void setObjectProperty(String name, Object value) throws JMSException; 25 .... 26 }
5. Message selectors.
a) Message selectors allow a JMSclient to specify which messages it wants to receive from a destination based on values in message headers.
b) Selectors are conditional expressions defined using a subset of SQL92.
6. Message body - JMS defines six Java types for the message body, also known as the payload.
a) Message - The base message type. Used to send a message with no payload, only headers and properties. Typically used for simple event notification.
b) TextMessage.
c) MapMessage - Uses a set of name/value pairs as its payload. The names are of type String and the values are a Java primitive type.
d) BytesMessage.
e) StreamMessage.
f) ObjectMessage - Used to hold a serializable Java object as its payload.
JMS domains
1. The point-to-point domain.
a) The point-to-point (PTP) messaging domain uses destinations known as queues.
b) Each message received on the queue is delivered once and only once to a single consumer.
c) Through the use of queues, messages are sent and received either synchronously or asynchronously.
2. The publish/subscribe domain.
a) The publish/subscribe (pub/sub) messaging domain uses destinations known as topics. Publishers send messages to the topic and subscribers register to receive messages from the topic.
b) Any messages sent to the topic are automatically delivered to all subscribers.
c) Much the same as PTP messaging, subscribers register to receive messages from the topic either synchronously or asynchronously.
3. Request/reply messaging.
a) Although the JMS spec doesn’t define request/reply messaging as a formal messaging domain, it does provide some message headers and a couple of convenience classes for handling basic request/reply messaging.
b) Request/reply messaging is an asynchronous back-and-forth conversational pattern utilizing either the PTP domain or the pub/sub domain through a combination of the JMSReplyTo and JMSCorrelationID message headers and temporary destinations. The JMSReplyTo specifies the destination where a reply should be sent, and the JMSCorrelationID in the reply message specifies the JMSMessageID of the request message.
c) The convenience classes for handling basic request/reply are the QueueRequestor and the TopicRequestor. These classes provide a request() method that sends a request message and waits for a reply message through the creation of a temporary destination where only one reply per requestis expected.
Administered objects
1. Administered objects contain provider-specific JMSconfiguration information and are supposed to be created by a JMSadministrator.
2. The JMS spec defines two types of administered objects: ConnectionFactoryand Destination.
Connection factory
1. JMS clients use the ConnectionFactory object to create connections to a JMS provider.
2. JMS connections are used by JMS clients to create javax.jms.Session objects that represent an interaction with the JMS provider.
Destination
The Destination object encapsulates the provider-specific address to which messages are sent and from which messages are consumed.
ActiveMQ message storage
How are messages stored by ActiveMQ?
1. Messages sent to queues and topics are stored differently.
2. Queue - Storage for queues is straightforward-messages are basically stored in first in, first out order (FIFO). Only when that message has been consumed and acknowledged can it be deleted from the broker’s message store.
3. Topic - In order to save storage space, only one copy of a message is stored by the broker. A durable subscriber object in the store maintains a pointer to its next stored message and dispatches a copy of it to its consumer. The message store is implemented in this manner because each durable subscriber could be consuming messages at different rates or they may not all be running at the same time. Also, because every message can potentially have many consumers, a message can’t be deleted from the store until it’s been successfully delivered to every interested durable subscriber.
The KahaDB message store
1. The recommended message store for general-purpose messages since ActiveMQ version 5.3 is KahaDB.
2. This is a file-based message store that combines a transactional journal, for reliable message storage and recovery, with good performance and scalability.
3. Enable the KahaDB store for ActiveMQ.
<broker brokerName="broker" persistent="true" useShutdownHook="false"> ... <persistenceAdapter> <kahaDB directory="activemq-data" journalMaxFileLength="16mb"/> </persistenceAdapter> ... </broker>
The AMQ message store
1. The AMQ message store, like KahaDB, is a combination of a transactional journal for reliable persistence (to survive system crashes) and high-performance indexes, which makes this store the best option when message throughput is the main requirement for an application.
2. But because it uses two separate files for every index, and there’s an index per destination, the AMQ message store shouldn’t beused if you intend to use thousands of queues per broker. Also, recovery can be slow if the ActiveMQ broker isn’t shut down cleanly, because all the indexes need to be rebuilt.
The JDBC message store
1. The most common reason why so many organizations choose the JDBC message store is because they already have expertise administering relational databases.
2. JDBC persistence is definitely not superior in performance to the aforementioned message store implementations.
3. Databases supported by the JDBC message store:
a) Apache Derby
b) MySQL
c) PostgreSQL
d) Oracle
e) SQL Server
f) Sybase
g) Informix
h) MaxDB
The memory message store
1. The memory message store holds all persistent messages in memory.
2. No active caching is involved, so you have to be careful that both the JVM and the memory limits you set for the broker are large enough to accommodate all the messages that may exist in this message store at one time.
Caching messages in the broker for consumers
1. The ActiveMQ message broker caches messages in memory for every topic that’s used. The only types of topics that aren’t supported are temporary topics and ActiveMQ advisory topics. Caching of messages in thisway isn’t handled for queues, as the normal operation of a queue is to hold every message sent to it.
2. Messages that are cached by the broker are only dispatched to a topic consumer if the consumer is retroactive, and never to durable topic subscribers.
Code examples
Publish/subscribe messaging
1. The first use case revolves around a stock portfolio use case for demonstrating publish/subscribe messaging.
2. This example is simple and utilizes a Publisher class for sending stock price messages to a topic, as well as a Consumer class for registering a Listener class to consume messages from topics in an asynchronous manner.
3. Publisher.java
1 import java.util.Hashtable; 2 import java.util.Map; 3 4 import javax.jms.Connection; 5 import javax.jms.ConnectionFactory; 6 import javax.jms.Destination; 7 import javax.jms.JMSException; 8 import javax.jms.MapMessage; 9 import javax.jms.Message; 10 import javax.jms.MessageProducer; 11 import javax.jms.Session; 12 13 import org.apache.activemq.ActiveMQConnectionFactory; 14 import org.apache.activemq.command.ActiveMQMapMessage; 15 16 public class Publisher { 17 18 protected int MAX_DELTA_PERCENT = 1; 19 protected Map<String, Double> LAST_PRICES = new Hashtable<String, Double>(); 20 protected static int count = 10; 21 protected static int total; 22 23 protected static String brokerURL = "tcp://centos1:61616"; 24 protected static transient ConnectionFactory factory; 25 protected transient Connection connection; 26 protected transient Session session; 27 protected transient MessageProducer producer; 28 29 public Publisher() throws JMSException { 30 factory = new ActiveMQConnectionFactory(brokerURL); 31 connection = factory.createConnection(); 32 try { 33 connection.start(); 34 } catch (JMSException jmse) { 35 connection.close(); 36 throw jmse; 37 } 38 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 39 producer = session.createProducer(null); 40 } 41 42 public void close() throws JMSException { 43 if (connection != null) { 44 connection.close(); 45 } 46 } 47 48 public static void main(String[] args) throws JMSException { 49 String[] stocks = { "IONA", "JAVA" }; 50 Publisher publisher = new Publisher(); 51 while (total < 1000) { 52 for (int i = 0; i < count; i++) { 53 publisher.sendMessage(stocks); 54 } 55 total += count; 56 System.out.println("Published '" + count + "' of '" + total + "' price messages"); 57 try { 58 Thread.sleep(1000); 59 } catch (InterruptedException x) { 60 } 61 } 62 publisher.close(); 63 } 64 65 protected void sendMessage(String[] stocks) throws JMSException { 66 int idx = 0; 67 while (true) { 68 idx = (int)Math.round(stocks.length * Math.random()); 69 if (idx < stocks.length) { 70 break; 71 } 72 } 73 String stock = stocks[idx]; 74 Destination destination = session.createTopic("STOCKS." + stock); 75 Message message = createStockMessage(stock, session); 76 System.out.println("Sending: " + ((ActiveMQMapMessage)message).getContentMap() + " on destination: " + destination); 77 producer.send(destination, message); 78 } 79 80 protected Message createStockMessage(String stock, Session session) throws JMSException { 81 Double value = LAST_PRICES.get(stock); 82 if (value == null) { 83 value = new Double(Math.random() * 100); 84 } 85 86 // lets mutate the value by some percentage 87 double oldPrice = value.doubleValue(); 88 value = new Double(mutatePrice(oldPrice)); 89 LAST_PRICES.put(stock, value); 90 double price = value.doubleValue(); 91 92 double offer = price * 1.001; 93 94 boolean up = (price > oldPrice); 95 96 MapMessage message = session.createMapMessage(); 97 message.setString("stock", stock); 98 message.setDouble("price", price); 99 message.setDouble("offer", offer); 100 message.setBoolean("up", up); 101 return message; 102 } 103 104 protected double mutatePrice(double price) { 105 double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT) - MAX_DELTA_PERCENT; 106 107 return price * (100 + percentChange) / 100; 108 } 109 110 }
4. Consumer.java
1 import javax.jms.Connection; 2 import javax.jms.ConnectionFactory; 3 import javax.jms.Destination; 4 import javax.jms.JMSException; 5 import javax.jms.MessageConsumer; 6 import javax.jms.Session; 7 8 import org.apache.activemq.ActiveMQConnectionFactory; 9 10 public class Consumer { 11 12 private static String brokerURL = "tcp://centos1:61616"; 13 private static transient ConnectionFactory factory; 14 private transient Connection connection; 15 private transient Session session; 16 17 public Consumer() throws JMSException { 18 factory = new ActiveMQConnectionFactory(brokerURL); 19 connection = factory.createConnection(); 20 connection.start(); 21 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 22 } 23 24 public void close() throws JMSException { 25 if (connection != null) { 26 connection.close(); 27 } 28 } 29 30 public static void main(String[] args) throws JMSException { 31 String[] stocks = { "IONA", "JAVA" }; 32 Consumer consumer = new Consumer(); 33 for (String stock : stocks) { 34 Destination destination = consumer.getSession().createTopic("STOCKS." + stock); 35 MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination); 36 messageConsumer.setMessageListener(new Listener()); 37 } 38 } 39 40 public Session getSession() { 41 return session; 42 } 43 44 }
5. Listener.java
1 import java.text.DecimalFormat; 2 3 import javax.jms.MapMessage; 4 import javax.jms.Message; 5 import javax.jms.MessageListener; 6 7 public class Listener implements MessageListener { 8 9 public void onMessage(Message message) { 10 try { 11 MapMessage map = (MapMessage)message; 12 String stock = map.getString("stock"); 13 double price = map.getDouble("price"); 14 double offer = map.getDouble("offer"); 15 boolean up = map.getBoolean("up"); 16 DecimalFormat df = new DecimalFormat( "#,###,###,##0.00" ); 17 System.out.println(stock + "\t" + df.format(price) + "\t" + df.format(offer) + "\t" + (up?"up":"down")); 18 } catch (Exception e) { 19 e.printStackTrace(); 20 } 21 } 22 23 }
Point-to-point messaging
1. The second use case focuses on job queues to illustrate point-to-point messaging.
2. This example uses a Producer class to send job messages to a job queue and a Consumer class for registering a Listener class to consume messages from queues in an asynchronous manner.
3. Publisher.java
1 import javax.jms.Connection; 2 import javax.jms.ConnectionFactory; 3 import javax.jms.Destination; 4 import javax.jms.JMSException; 5 import javax.jms.Message; 6 import javax.jms.MessageProducer; 7 import javax.jms.ObjectMessage; 8 import javax.jms.Session; 9 10 import org.apache.activemq.ActiveMQConnectionFactory; 11 12 public class Publisher { 13 14 private static String brokerURL = "tcp://centos1:61616"; 15 private static transient ConnectionFactory factory; 16 private transient Connection connection; 17 private transient Session session; 18 private transient MessageProducer producer; 19 20 private static int count = 10; 21 private static int total; 22 private static int id = 1000000; 23 24 private String jobs[] = new String[]{"suspend", "delete"}; 25 26 public Publisher() throws JMSException { 27 factory = new ActiveMQConnectionFactory(brokerURL); 28 connection = factory.createConnection(); 29 connection.start(); 30 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 31 producer = session.createProducer(null); 32 } 33 34 public void close() throws JMSException { 35 if (connection != null) { 36 connection.close(); 37 } 38 } 39 40 public static void main(String[] args) throws JMSException { 41 Publisher publisher = new Publisher(); 42 while (total < 1000) { 43 for (int i = 0; i < count; i++) { 44 publisher.sendMessage(); 45 } 46 total += count; 47 System.out.println("Published '" + count + "' of '" + total + "' job messages"); 48 try { 49 Thread.sleep(1000); 50 } catch (InterruptedException x) { 51 } 52 } 53 publisher.close(); 54 55 } 56 57 public void sendMessage()
