[ InAction Note ] Ch5. Message Storage - The JDBC message store

Preface: 
The flexibility of the ActiveMQ pluggable message store API allows for many different implementation choices. The oldest and more common store implementation uses JDBC for messaging persistence. The most common reason why so many organizations choose the JDBC message store is because they already have expertise administering relational databases. JDBC persistence is definitely not superior in performance to the aforementioned message store implementations. The fact of the matter is that many businesses have invested in the use of relational databases so they prefer to make full use of them. 

But the use of a shared database is particularly useful for making a redundant master/slave topology out of multiple brokers. When a group of ActiveMQ brokers is configured to use a shared database, they’ll all try to connect and grab a lock in the lock table, but only one will succeed and become the master. The remaining brokers will be slaves, and will be in a wait state, not accepting client connections until the master fails. This is a common deployment scenario for ActiveMQ, which will be covered in more detail in chapter 10. 

When using the JDBC message store, the default JDBC driver used in ActiveMQ is Apache Derby. But many other relational databases are supported. 

Databases supported by the JDBC message store: 
This isn’t an exhaustive list, the JDBC store has been shown to operate with the following relational databases: 

* Apache Derby
* MySQL
* PostgreSQL
* Oracle
* SQL Server
* Sybase
* Informix
* MaxDB

Some users prefer to use a relational database for message persistence simply because of the ability to query the database to examine messages. The following sections will discuss this topic. 

The JDBC message store schema: 
The JDBC message store uses a schema consisting of three tables. Two of the tables are used to hold messages, and the third is used as a lock table to ensure that only one ActiveMQ broker can access the database at one time. Here’s a detailed breakdown of these tables. 

The message table, shown in table 5.3, is by default named ACTIVEMQ_MSGS and is defined as follows. 
 

Messages are broken down and stored into the ACTIVEMQ_MSGS table for both queues and topics. 

There’s a separate table for holding durable subscriber information and an ID to the last message the durable subscriber received. This information is held in the ACTIVEMQ_ACKS table, which is shown in table 5.4. 
 

For durable subscribers, the LAST_ACKED_ID sequence is used as a simple pointer into the ACTIVEMQ_MSGS and enables messages for a particular durable subscriber to be easily selected from the ACTIVEMQ_MSGS table. 

The lock table, called ACTIVEMQ_LOCK, is used to ensure that only one ActiveMQ broker instance can access the database at one time. If an ActiveMQ broker can’t grab the database lock, that broker won’t initialize fully, and will wait until the lock becomes free, or it’s shut down. The table structure of the lock table is defined in table 5.5. 
 

Configuring the JDBC message store: 
Configuring the default JDBC message store is straightforward and the default JDBC store uses Apache Derby in the broker configuration as shown: 

The preceding configuration sets the persistence adaptor for the ActiveMQ broker to be the JDBC message store (which uses Apache Derby by default) and sets the data directory to be used by the embedded Apache Derby instance. 


One of the key properties on the JDBC persistence adapter (the interface onto the JDBC message store) is the dataSource property. This property defines a factory from which connections to a relational database are created. Configuring the dataSource object enables the JDBC persistence adaptor to use physical databases other than the default. Here’s an example of an ActiveMQ configuration for the JDBC message store using the MySQL database: 

The preceding example uses the Apache Commons DBCP BasicDataSource to wrap the MySQL JDBC driver for connection pooling. In the example, the driverClassName is the name of the JDBC driver to use. Some properties that you can configure are passed directly to the database driver itself. For example, maxActive is a property for the MySQL database connector, which tells the database how many active connections to hold open at one time. 


Just as a point of comparison, here’s an example of a configuration to use the Oracle database: 

This example uses the Apache Commons DBCP BasicDataSource to wrap the Oracle JDBC driver for connection pooling. 


Using the JDBC message store with the ActiveMQ journal: 
Though the performance of the JDBC message store isn’t wonderful, it can be improved through the use of the ActiveMQ journal. The journal ensures the consistency of JMS transactions. Because it incorporates fast message writes with caching technology, it can significantly improve the
performance of the ActiveMQ broker. 

Here’s an example configuration using the journal with JDBC (aka journaled JDBC). In this case, Apache Derby is being used. 

The journal can be used with any JDBC datasource, but it’s important to know when it should and shouldn’t be used. 


The journal offers considerable performance advantages over the use of a standard JDBC message store, especially when the JDBC database is co-located on the same machine as the ActiveMQ broker. The only time when it’s not possible to use the journal is in a shared database master/slave configuration. Because messages from the master may be stored locally in the journal before they’ve been committed to the database, using the journal in this configuration could lead to lost messages if the master failed because the journal isn’t replicated.

[ InAction Note ] Ch5. Message Storage - The AMQ message store

Preface: 
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. But because it uses two separate files for every index, and there’s an index per destination, the AMQ message store shouldn’t be used if you intend to use thousands of queues per broker. Also, recovery can be slow if the ActiveMQ broker isn’t shut down cleanly. 

The AMQ message store internals: 
The main components of the AMQ message store are similar to that of the KahaDB message store, in that there’s a cache, message data logs, and a reference store for accessing the data logs in order. Figure 5.5 provides a high-level diagram of the AMQ message store. 
 

The diagram provides a view of the three distinct parts of the AMQ message store: 
- The data logs: These act as a message journal. 
- The cache: This holds messages for fast retrieval in memory after they’ve been written to the data logs. 
- The reference store: This holds references to the messages in the journal that are indexed by their message ID. 

The AMQ message store directory structure: 
When you start ActiveMQ with the AMQ message store configured, a directory will automatically be created in which the persistent messages are held. The AMQ message store directory contains subdirectories for all the brokers that are running on the machine. For this reason it’s strongly recommended that each broker use a unique name. In the default configuration for ActiveMQ, the broker name is localhost, which needs to changed to something unique. This directory structure is represented in figure 5.6—the AMQ store directory structure. 

The following directories and files can be found inside the data directory of an ActiveMQ broker: 
- A lock file 

Ensures that only one broker can access this data at any given time. The lock is commonly used for hot standby purposes where more than one bro-
ker with the same name will exist on the same system.

- A temp-storage directory 

Used for storing nonpersistent messages that can no longer be stored in broker memory. These messages are typically awaiting delivery to a slow consumer.

- The kr-store 

The directory structure used by the reference (index) part of the AMQ message store. It uses the Kaha reference store by default (Kaha is part of the ActiveMQ core library) to index and store references to messages in the data logs. There are two distinct parts to the kr-store:
* The data directory

Contains the indexes and collections used to reference the messages held in the data logs. This data directory is deleted and rebuilt as part of recovery, if the broker hasn’t shut down cleanly. You can force recovery by manually deleting this directory before starting the broker.

* The state directory

Holds information about durable topic consumers. The journal itself doesn’t hold information about consumers, so when it’s recovered it has to retrieve information about the durable subscribers first to accurately rebuild its database.

- The journal directory 

Contains the data files for the data logs, and a data-control file that holds some meta information. The data files are reference counted, so when all the contained messages are delivered, a data file can be deleted or archived.

- The archive directory 

Exists only if archiving is enabled. Its default location can be found next to the journal. It makes sense to use a separate partition or disk. The archive is used to store data logs from the journal directory, which are moved here instead of being deleted. This makes it possible to replay messages from the archive at a later point. To replay messages, move the archived data logs (or a subset) to a new journal directory and start a new broker pointed to the location of this directory. It’ll automatically replay the data logs in the journal.

Configuring the AMQ message store: 
The AMQ store configuration allows the user to change its basic behaviour around indexing, checkpoint intervals, and the size of the journal data files. These items and many more can be customized through the use of properties. The key properties for the AMQ store are shown in table 5.2. 

Table 5.2 Configuration properties for the AMQ message store 

Below is an example of using the properties from table 5.2 in an ActiveMQ XML configuration file: 

This is but a small example of a customized configuration for the AMQ store using the available properties.

[ InAction Note ] Ch5. Message Storage - The KahaDB message store

Preface: 
The JMS specification supports two types of message delivery: persistent and non-persistent. A message delivered with the persistent delivery property must be logged to stable storage. For nonpersistent messages, a JMS provider must make best efforts to deliver the message, but it won’t be logged to stable storage. 

ActiveMQ supports both of these types of message delivery and can also be configured to support message recovery, an in-between state where messages are cached in memory. ActiveMQ supports a pluggable strategy for message storage and provides storage options for in-memory, file-based, and relational databases. 

Persistent messages are used if you want messages to always be available to a message consumer after they’ve been delivered to the broker, even if that consumer isn’t running when the message was sent. Once a message has been consumed and acknowledged by a message consumer, it’s typically deleted from the broker’s message store. 

Nonpersistent messages are typically used for sending notifications or real-time data. You should use nonpersistent messages when performance is critical and guaranteed delivery of the message isn’t required. 

How are messages stored by ActiveMQ? 
It’s important to gain some basic knowledge of the storage mechanisms for messages in an ActiveMQ message store. This will aid in configuration and provide an awareness of what takes place in the ActiveMQ broker during the delivery of persistent messages. Messages sent to queues and topics are stored differently, because there are some storage optimizations that can be made with topics that don’t make sense with queues, as we’ll explain. 

Storage for queues is straightforward—messages are basically stored in first in, first out order (FIFO). See figure 5.1 for a depiction of this. One message is dispatched to a single consumer at a time. Only when that message has been consumed and acknowledged can it be deleted from the broker’s message store. 
 

For durable subscribers to a topic, each consumer gets a copy of the message. 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 as shown in figure 5.2. 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: 
The recommended message store for general-purpose messages since ActiveMQ version 5.3 is KahaDB. This is a file-based message store that combines a transactional journal, for reliable message storage and recovery, with good performance and scalability. 

The KahaDB store is a file-based, transactional store that’s been tuned and designed for the fast storage of messages. The aim of the KahaDB store is to be easy to use and as fast as possible. Its use of a file-based message database means there’s no prerequisite for a third-party database. This message store enables ActiveMQ to be downloaded and running in literally minutes. In addition, the structure of the KahaDB store has been streamlined especially for the requirements of a message broker. 

The KahaDB message store uses a transactional log for its indexes and only uses one index file for all its destinations. It’s been used in production environments with 10,000 active connections, each connection having a separate queue. The configurability of the KahaDB store means that it can be tuned for most usage scenarios, from high throughput applications (for example, trading platforms), to storing large amounts of messages (for example, GPS tracking). 

To enable the KahaDB store for ActiveMQ, you need to configure the  element in the activemq.xml configuration file. Here’s a minimal configuration for the KahaDB message store: 

If you want to embed an ActiveMQ broker inside an application, the message store can also be configured programmatically. Here’s an example of a programmatic configuration for KahaDB: 

 

Although the example seems small, it’s enough to create an ActiveMQ broker using the KahaDB message store and listen for ActiveMQ clients connecting over TCP. For more information about embedding ActiveMQ, see chapter 8. 

The KahaDB message store internals 
The KahaDB message store is the fastest of all the provided message store implementations. Its speed is the result of the combination of a fast transactional journal comprised of data log files, the highly optimized indexing of message IDs, and in-memory message caching. Figure 5.3 provides a high-level diagram of the KahaDB message store. 
 

The diagram provides a view of the three distinct parts of the KahaDB message store including the following: 
- The data logs 

The data logs act as a message journal, which consists of a rolling log of messages and commands (such as transactional boundaries and message deletions) stored in data files of a certain length. When the maximum length of the currently used data file has been reached, a new data file is created. All the messages in a data file are reference counted, so that once every message in that data file is no longer required, the data file can be removed or archived. In the data logs, messages are only appended to the end of the current data file, so storage is fast.

- The cache 

The cache holds messages temporarily if there are active consumer(s) for the messages. If there are active consumers, messages are dispatched at the same time they’re scheduled to be stored. If messages are acknowledged in time, they don’t need to be written to disk.

- The BTree indexes 

The BTree indexes hold references to the messages in the data logs that are indexed by their message ID. The indexes maintain the FIFO data structure for queues and the durable subscriber pointers to their topic messages. The redo log is used only if the ActiveMQ broker hasn’t shut down cleanly, and are used to insure the integrity of the BTree index is maintained.

The KahaDB message store directory structure: 
When you start an ActiveMQ broker configured to use a KahaDB store, a directory will automatically be created in which the persistent messages are stored. This directory structure is shown in figure 5.4. 
 

Inside of the KahaDB directory, the following directory and file structures can be found: 
- db log files 

KahaDB stores messages into data log files named db-.log of a predefined size. When a data log is full, a new one will be created, and the log number incremented. When there are no more references to any of the messages in the data log file, it’ll be deleted or archived.

- archive directory 

This exists only if archiving is enabled. The archive is used to store data logs that are no longer needed by KahaDB, making it possible to replay messages from the archived data logs at a later point. If archiving isn’t enabled (the default), data logs that are no longer in use are deleted from the file system.

- db.data 

This file contains the persistent BTree indexes to the messages held in the message data logs.

- db.redo 

This is the redo file, used for recovering the BTree indexes if the KahaDB message store starts after a hard stop.

Configuring the KahaDB message store: 
The KahaDB message store can be configured in the activemq.xml file. Its configuration options control the different tuning parameters (table 5.1): 
 
Table 5.1 Configuration options available for the KahaDB message store 

ActiveMQ provides a pluggable API for message stores, and there are three additional implementations to KahaDB that are shipped with ActiveMQ: 

* The AMQ message store—A file-based message store designed for performance
* The JDBC message store—A message store based on JDBC
* The Memory message store—A memory-based message store

We’ll look at the use cases and configuration for these additional message stores in the next three sections.

[ InAction Note ] Ch4. Connecting ActiveMQ - Network connectors (2)

Dynamic networks: 
Thus far we’ve seen how to set up broker networks and connect to them by explicitly specifying broker URIs (both transport and network connectors). As you’ll see in this section, ActiveMQ implements several mechanisms that can be used by brokers and clients to find each other and establish necessary connections. 

MULTICAST CONNECTOR 
IP multicast is a network technique used for easy transmission of data from one source to a group of interested receivers (one-to-many communications) over an IP network. One of the fundamental concepts of IP multicast is the so-called group address. The group address is an IP address in the range of 224.0.0.0 to 239.255.255.255 used by both sources and receivers. Sources use this address as a destination for their data, whereas receivers use it to express their interest in data from that group. 

When IP multicast is configured, ActiveMQ brokers use the multicast protocol to advertise their services and locate the services of other brokers for the purpose of creating networks of brokers. Clients, on the other hand, use multicast to locate brokers and establish a connection with them. This section discusses how brokers use multicast; the use of multicast by a client will be discussed later. 

The URI syntax for the multicast protocol is as follows: 

multicast://ipadaddress:port?key=value

This is no different than the previous URIs with the exception of the scheme portion. 

Here’s a snippet from the default ActiveMQ configuration that makes use of multicast: 

In the example, the group name default is used instead of a specific IP address. There are two important things achieved with this configuration snippet. First, the transport connector’s discoveryUri attribute is used to advertise this transport’s URI on the default group. All clients interested in finding an available broker would use this connector. This will be demonstrated in the following section. 

Next, the uri attribute of the network connector is used to search for available brokers and to create a network with them. In this case, the broker acts like a client and uses multicast for lookup purposes. You can find a complete reference of this protocol at the ActiveMQ website (http://mng.bz/14yJ). 

EXAMPLE USE OF THE MULTICAST PROTOCOL 
The multicast protocol is somewhat different from the TCP protocol. The difference is the automatic discovery of other brokers instead of using a static list of brokers. Use of the multicast protocol is common where brokers are added and removed frequently, and in cases where brokers may have their IP addresses changed frequently. In these cases, instead of reconfiguring each broker manually for every change, it’s often easier to utilize a discovery protocol. 

One disadvantage to using the multicast protocol is that discovery is automatic. If there are brokers that you don’t want to be automatically added to a given group, you must be careful in setting up the initial configuration of the broker network. Careful segmentation of broker networks is important, as you don’t want messages to wind up in a broker network where they don’t belong. Another disadvantage of the multicast protocol is that it can be excessively chatty on the network. For this reason, many network administrators won’t allow its use. Please check with your network administrator before taking the time to configure a network using the multicast protocol. 

DISCOVERY PROTOCOL 
The discovery transport connector is on the client side of the ActiveMQ multicast functionality. This protocol is basically the same as the failover protocol in its behavior. The only difference is that it’ll use multicast to discover available brokers and randomly choose one to connect to. The syntax of this protocol is: 

discovery:(discoveryAgentURI)?key=value

Its complete reference could be found at the ActiveMQ website (http://mng.bz/96wI). 

Using the multicast broker configuration explained earlier, you can run the broker with the following command: 

${ACTIVEMQ_HOME}/bin/activemq console \
xbean:src/main/resources/org/apache/activemq/book/ch4/activemq-multicast.xml

Once the broker is started, run the stock portfolio publisher with the following command: 

$ mvn -e exec:java \
-Dexec.mainClass=org.apache.activemq.book.ch4.Publisher \
-Dexec.args="discovery:(multicast://default) CSCO ORCL"

You’ll notice the following log messages at the application startup: 
 
These messages tell you that the publisher client has successfully used multicast to dis- cover and connect to the local broker. 

PEER PROTOCOL 
As we’ve seen before, networks of brokers and embedded brokers are useful concepts that allow you to fit brokers to your infrastructure needs. Of course, it’s theoretically possible to create networks of embedded brokers, but this would be quite cumbersome to configure manually. This is why ActiveMQ provides the peer transport connector, as it allows you to more easily network embedded brokers. The peer connector is a utility transport that is a superset of a VM connector that creates a peer-to-peer network of embedded brokers. 

The URI syntax of this protocol is as follows: 

peer://peergroup/brokerName?key=value

You can find its complete reference at the ActiveMQ website (http://mng.bz/bIaH). 

When started with the peer protocol URI, the application will automatically start an embedded broker (just as was the case with the VM protocol), but will also configure the broker to establish network connections to other brokers in the local network with the same group name. 

Let’s walk through a demonstration of this using the stock portfolio example with the peer protocol. In this case, both the publisher and the consumer will use their own embedded brokers that will be networked automatically. Figure 4.7 provides a better perspective of this solution. 
 

Advise the stock portfolio publisher to create its own embedded broker using group1 like this: 

$ mvn -e exec:java -Dexec.mainClass=org.apache.activemq.book.ch4.Publisher \
-Dexec.args="peer://group1 CSCO ORCL"
...

Also advise the stock portfolio consumer to create its own embedded broker using group1 like this: 

$ mvn -e exec:java -Dexec.mainClass=org.apache.activemq.book.ch4.Consumer \
-Dexec.args="peer://group1 CSCO ORCL"
...

The two commands start two embedded brokers (one for each application) and create a peer-to-peer broker network named group1 between these two brokers. All messages sent to one broker will be available in the other broker as well as any other brokers that might join group1. Note that the overall system operates as if these two applications were using the same centralized broker. 

EXAMPLE USE OF THE PEER PROTOCOL 
Consider an application that resides on the laptop of a field sales representative who often disconnects from the company network but still needs the application to run successfully in a disconnected mode. This is a common scenario where the client application needs to continue working regardless of whether the network is available. This is a case where the peer protocol can be utilized for an embedded broker to allow the application on the laptop to keep running successfully. In reality, while in disconnected mode, the application is simply sending messages to the local broker, where they’re queued up to be sent at a later time when the network is available again.The sales rep can still log client calls, visits, and so on while the laptop is disconnected from the network. When the laptop is again connected to the network, all of the queued messages will be sent along based on the demand from consuming clients. 

FANOUT CONNECTOR 
Fanout is another utility connector used by clients to simultaneously connect to multiple brokers and replicate operations to those brokers. The URI syntax as follows: 

fanout:(fanoutURI)?key=value

You can find its complete reference at the ActiveMQ website (http://mng.bz/J7i0). 

The fanoutURI can utilize either a static URI or a multicast URI. Consider the following example: 

fanout:(static:(tcp://host1:61616,tcp://host2:61616,tcp://host3:61616))

In figure 4.8, the client will try to connect to three brokers statically defined using the static protocol 
 

The same effect could be accomplished by simply using the following URI: 

fanout:(multicast://default)

This assumes that the brokers are configured to use multicast to advertise their transport connectors. 

By default, the fanout protocol will wait until it connects to at least two brokers and won’t replicate commands to queues (only topics). Both of these features are, of course, configurable with appropriate transport options. 

Finally, there are a couple of things you should be aware of if you plan to use the fanout protocol. First of all, it’s not recommended for consuming messages. Its only purpose is to produce messages to multiple brokers. Also, if the brokers you’re using are in the same network of brokers, it’s likely that certain consumers will receive duplicate messages. So basically, the fanout protocol is only recommended for publishing messages to multiple nonconnected brokers. 

With the fanout protocol, we come to the end of the discussion on networks of brokers and network connectors. For reference purposes, table 4.2 provides a summary of all the protocols covered in this section.