Shulou(Shulou.com)06/02 Report--

This article will explain in detail how to implement message storage in ActiveMQ. The editor thinks it is very practical, so I share it with you as a reference. I hope you can get something after reading this article.

I. queue and topic

Overview

ActiveMQ supports not only persistent and non-persistent, but also message recovery (recovery)

PTP

The storage mode of Queue is very simple, which is a FIFO (first-in, first-out) Queue

PUB/SUB

For persistent subscription topics, each consumer will get a copy of the message

Effective message storage

ActiveMQ provides a plug-in message storage, which is similar to the multipoint propagation of messages. It mainly implements the following:

1:AMQ message store is a file-based storage method that used to be the default message store.

2:KahaDB message Storage 1 provides capacity enhancement and resilience, and is now the default storage mode.

3:JDBC message store a message based on JDBC storage

4:Memory message store A memory-based message store

II. KahaDB

Overview of KahaDB Message Store

KahaDB is currently the default storage method, which can be used in any scenario to improve performance and resilience. The message store uses a transaction log and only one index file to store all its addresses.

KahaDB is a solution for message persistence that optimizes typical message usage patterns. In Kaha, the data is appended to the datalogs. When the number in the log file is no longer needed, the log file is discarded.

KahaDB basic configuration example

The available properties are:

The path where 1:director:KahaDB is stored. Default is activemq-data.

2:indexWriteBatchSize: the number of index page written to disk in bulk. Default is 1000.

3:indexCacheSize: number of cached index page in memory. Default is 10000.

4:enableIndexWriteAsync: whether to write the index asynchronously. Default is false.

5:journalMaxFi1eLength: sets the size of each message data log. The default is 32m.

6:enab1eJournalDiskSyncs: sets whether to ensure that every content without transaction is written to disk synchronously, which is required when JMS is persisted. Default is true.

7:cleanupInterval: the time before the message is deleted after checking that it is no longer in use. Default is 30000.

Interval between 8:checkpointInterval:checkpoint. Default is 5000.

9:ignoreMissingJournalfiles: whether to ignore lost message log files, default false

10:checkForCourruptJournalFiles: upon startup, the message file will be verified for corruption. The default is false.

11:checksumJournalFiles: whether to provide checksum for each message log file, default false

12:archiveDataLogs: whether to move files to a specific path instead of deleting them, default false

13:directoryArchive: defines the path to which the data log is moved after the message has been consumed. Default is null.

14:databaseLockedWaitDelay: the waiting time for getting the database lock. Default is 10000.

15:maxAsyncJobs: sets the maximum asynchronous message queue that can be stored. The default is 10000, which can be set to the same value as concurrent MessageProducers.

16:concurrentStoreAndDispatchTransactions: whether to distribute the message to the client, while the transaction stores the message. Default true

17:concurrentStoreAndDispatchTopics: whether to distribute Topic messages to the client and store them at the same time. Default true

18:concurrentStoreAndDispatchQueues: whether to distribute queue messages to the client and store them at the same time. Default true

An example of using Broker and KahaDB embedded in Java

Public class EmbeddedBrokerUsingKahaDBStoreExample {public BrokerService createEmbeddedBroker () throws Exception {BrokerService broker = new BrokerService (); File dataFileDir = new File ("target/amq-in-action/kahadb"); KahaDBStore kaha = new KahaDBStore (); kaha.setDirectory (dataFileDir); kaha.setJournalMaxFileLength (1024 to 1000); kaha.setIndexWriteBatchSize (100); kaha.setEnableIndexWriteAsync (true); broker.setPersistenceAdapter (kaha); broker.addConnector ("tcp://localhost:61616"); broker.start (); return broker;}}

III. AMQ

Overview of AMQ Message Store

AMQ Message Store is the default persistent storage of ActiveMQ5.0. It is a structure designed for fast message storage based on file and transaction storage. This structure carries on message interaction in the form of stream.

In this way, Messages is saved to data logs and indexed by reference store to improve access speed. Data logs consists of separate data log files, and the default file size is 32m. If the size of a message exceeds the size of the data log file, you can modify the configuration to increase the size of the data log file. If all messages in a data log file are successfully consumed, the data log file will be marked for deletion or archiving in the next round of cleanup.

AMQ Message Store configuration example

IV. JDBC

Use JDBC to persist messages (this step does not require manual scripting)

ActiveMQ supports using JDBC to persist messages. The predefined tables are as follows:

1: message table. By default, both ACTIVEMQ_MSGS,quue and topic are stored in it. The structure is as follows:

The 2:ACTIVEMQ_ACKS table stores the information of the persistent subscription and the message ID received by the last persistent subscription, which is structured as follows:

3: lock the table. The default is ACTIVEMQ_LOCK, which is used to ensure that only one Act broker instance can access the database at a certain time. The structure is as follows:

Example of a configuration that uses JDBC to persist messages

Note:

(1) the database needs to set the character set to latin1.

(2) you need to put the mysql-connector-java.jar package into the lib.

(3) three tables will appear after a successful startup.

Example

1. (queue mode):

Run the sender (there are three unreceived messages in the database):

Run the recipient (when the message is received successfully, the message in the database will also be deleted)

II. Persistence mode

After the message receiver has finished receiving the message, the message in the database is not deleted.

As shown in the figure:

JDBC Message Store with ActiveMQ Journal (log)

This method overcomes the shortcomings of JDBC Store, uses fast cache write technology, and greatly improves the performance. Examples of configurations are as follows:

The difference between JDBC Store and JDBC Message Store with ActiveMQ Journal

The performance of 1:jdbc with journal is better than jdbc.

2:jdbc is used for database sharing of master/slave schema

3:jdbc with journal cannot be used in master/slave mode

4: in general (non-cluster), jdbc with journal is recommended.

5. MMS

Memory Message Store

The main purpose of memory message storage is to store all persistent messages in memory. There is no dynamic cache here, so you must pay attention to setting the JVM and memory limits where broker is located.

Memory Message Store configuration example

An example of using Broker and Memory embedded in Java

Public void createEmbeddedBroker () throws Exception {BrokerService broker = new BrokerService (); broker.setPersistent (false); broker.addConnector ("tcp://localhost:61616"); broker.start ();} this is the end of the article on "how to implement message storage in ActiveMQ". I hope the above content can be of some help to you, so that you can learn more knowledge. if you think the article is good, please share it for more people to see.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.