activemq 数据持久化mysql

基础概念

ActiveMQ 是一个开源的消息中间件，用于在分布式系统中传递消息。它支持多种消息协议，如 AMQP、MQTT 等。数据持久化是将消息存储在磁盘上，以确保在消息代理重启后消息不会丢失。

MySQL 是一个关系型数据库管理系统，广泛用于数据存储和管理。将 ActiveMQ 的消息持久化到 MySQL 中，可以提高消息的可靠性和可恢复性。

优势

可靠性：消息持久化到 MySQL 可以确保消息在系统故障时不会丢失。
可恢复性：系统重启后，可以从 MySQL 中恢复消息。
灵活性：MySQL 提供了强大的查询和管理功能，便于对消息进行管理和审计。
扩展性：MySQL 可以轻松扩展以处理大量消息。

类型

ActiveMQ 支持多种持久化方式，包括：

KahaDB：ActiveMQ 默认的持久化方式，使用文件系统存储消息。
JDBC：使用 JDBC 将消息持久化到关系型数据库（如 MySQL）。
LevelDB：使用 LevelDB 存储消息，适用于高性能场景。

应用场景

金融系统：确保交易消息的可靠传递。
电子商务系统：确保订单消息的可靠传递。
物联网系统：确保设备状态和数据的可靠传递。

配置示例

以下是一个将 ActiveMQ 消息持久化到 MySQL 的配置示例：

1. 安装 MySQL

首先，确保你已经安装并配置好了 MySQL 数据库。

2. 创建数据库和表

CREATE DATABASE activemq;
USE activemq;

CREATE TABLE queue (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    message_id VARCHAR(255) NOT NULL,
    message TEXT NOT NULL,
    expiration BIGINT,
    inserted TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

3. 配置 ActiveMQ

编辑 ActiveMQ 的 conf/activemq.xml 文件，添加 JDBC 持久化配置：

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}">

    <destinationPolicy>
        <policyMap>
            <policyEntries>
                <policyEntry queue=">" producerFlowControl="true" optimizedDispatch="true" maxPageSize="200" maxBrowsePageSize="200" />
            </policyEntries>
        </policyMap>
    </destinationPolicy>

    <persistenceAdapter>
        <jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="true" />
    </persistenceAdapter>

    <systemUsage>
        <systemUsage>
            <memoryUsage>
                <memoryUsage percentOfJvmHeap="70" />
            </memoryUsage>
            <storeUsage>
                <storeUsage limit="10 gb"/>
            </storeUsage>
            <tempUsage>
                <tempUsage limit="5 gb"/>
            </tempUsage>
        </systemUsage>
    </systemUsage>

    <plugins>
        <simpleAuthenticationPlugin>
            <users>
                <authenticationUser username="admin" password="admin" groups="admins,publishers,consumers"/>
            </users>
        </simpleAuthenticationPlugin>
    </plugins>

    <transportConnectors>
        <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    </transportConnectors>

    <dataSources>
        <dataSource id="mysql-ds" driverClassName="com.mysql.jdbc.Driver" url="jdbc:mysql://localhost:3306/activemq" username="root" password="password" maxActive="20"/>
    </dataSources>

</broker>