ActiveMQ持久化消息與非持久化消息

需求描述：將ActiveMQ消息持久化到mySql\Oracle數(shù)據(jù)庫中 ;

環(huán)境描述：目前最新版本是ActiveMQ5.13.2，本文講述的實例是ActiveMQ5.9.0。

使用默認的持久化機制，我們不容易直接看到消息究竟是如何持久的。ActiveMQ提供的JDBC持久化機制，能夠將持久化信息存儲到數(shù)據(jù)庫。通過查看數(shù)據(jù)庫中ActiveMQ生成的表結構和存儲的數(shù)據(jù)，能夠幫助我們更好的了解消息的持久化機制。現(xiàn)在介紹如何配置activemq，將數(shù)據(jù)持久化到mysql中。

1.配置activeMQ需要的mySql數(shù)據(jù)源

為了能夠使用JDBC訪問mysql數(shù)據(jù)庫，顯然必須要配置消息服務器的數(shù)據(jù)庫源。在activemq\apache-activemq-5.9.0\conf\activemq.xml進行配置

<!-- MySQL DataSource -->
	<bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
		<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
		<property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>
		<property name="username" value="root"/>
		<property name="password" value="root"/>
		<property name="poolPreparedStatements" value="true"/>

    </bean>

在結點之后，增加數(shù)據(jù)源的配置

2.改變activeMQ默認的持久化方式

在activemq.xml中注釋掉默認的kahadb，使用jdbc持久化

<!--
        <persistenceAdapter>
            <kahaDB directory="${activemq.data}/kahadb"/>
        </persistenceAdapter>
		-->
		
		<persistenceAdapter>
			<jdbcPersistenceAdapter  dataSource="#mysql-ds"/>

        </persistenceAdapter>

3.設置JMS訪問連接協(xié)議類型

修改原來的協(xié)議連接為如下方式：

<transportConnectors>
	<transportConnector name="default" uri="tcp://localhost:61616"/>
</transportConnectors>

4.提供mysql的驅動程序

由于activeMQ消息服務器，沒有自帶mysql數(shù)據(jù)庫的驅動程序。我們需要手動將mysql驅動添加到消息服務器。

方法是將驅動拷貝到apache-activemq-5.9.0\lib\目錄下。

經(jīng)過上面的三步配置，我們重新啟動消息服務器，就可以發(fā)現(xiàn)activeMQ在數(shù)據(jù)庫中新建了3張表activemq_acks ，activemq_lock ，activemq_msgs 。

至此數(shù)據(jù)庫持久化完成。ActiveMQ持久化的中表結構是什么，表需要人工創(chuàng)建嗎?其實不需要，ActiveMQ會幫助我們生成的。只需要制定采用的數(shù)據(jù)庫名稱并，創(chuàng)建數(shù)據(jù)庫即可。以為為ActiveMQ采用MySQL5.7持久化產生的SQL語句：

持久化mysql數(shù)據(jù)庫的3張表;

activemq_acks：ActiveMQ的簽收信息。

activemq_lock:ActiveMQ的鎖信息。

activemq_msgs:ActiveMQ的消息的信息

5.activemq.xml全部配置信息

<!--
    Licensed to the Apache Software Foundation (ASF) under one or more
    contributor license agreements.  See the NOTICE file distributed with
    this work for additional information regarding copyright ownership.
    The ASF licenses this file to You under the Apache License, Version 2.0
    (the "License"); you may not use this file except in compliance with
    the License.  You may obtain a copy of the License at
    http://www.apache.org/licenses/LICENSE-2.0
    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
-->
<!-- START SNIPPET: example -->
<beans
  xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
 
    <!-- Allows us to use system properties as variables in this configuration file -->
    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <value>file:${activemq.conf}/credentials.properties</value>
        </property>
    </bean>
 
    <!-- Allows log searching in hawtio console -->
    <bean id="logQuery" class="org.fusesource.insight.log.log4j.Log4jLogQuery"
          lazy-init="false" scope="singleton"
          init-method="start" destroy-method="stop">
    </bean>
	
	<!-- MySql DataSource Sample Setup -->
  <bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
    <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
    <property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>
    <property name="username" value="root"/>
    <property name="password" value="root"/>
    <property name="poolPreparedStatements" value="true"/>
  </bean>
    
  <!-- Oracle DataSource Sample Setup -->
  <!--
  <bean id="oracle-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
    <property name="driverClassName" value="oracle.jdbc.driver.OracleDriver"/>
    <property name="url" value="jdbc:oracle:thin:@localhost:1521:AMQDB"/>
    <property name="username" value="scott"/>
    <property name="password" value="tiger"/>
    <property name="poolPreparedStatements" value="true"/>
  </bean>
  -->
 
    <!--
        The <broker> element is used to configure the ActiveMQ broker.
    -->
    <broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" >
 
        <destinationPolicy>
            <policyMap>
              <policyEntries>
                <policyEntry topic=">" >
                    <!-- The constantPendingMessageLimitStrategy is used to prevent
                         slow topic consumers to block producers and affect other consumers
                         by limiting the number of messages that are retained
                         For more information, see:
                         http://activemq.apache.org/slow-consumer-handling.html
                    -->
                  <pendingMessageLimitStrategy>
                    <constantPendingMessageLimitStrategy limit="1000"/>
                  </pendingMessageLimitStrategy>
                </policyEntry>
              </policyEntries>
            </policyMap>
        </destinationPolicy>
 
 
        <!--
            The managementContext is used to configure how ActiveMQ is exposed in
            JMX. By default, ActiveMQ uses the MBean server that is started by
            the JVM. For more information, see:
            http://activemq.apache.org/jmx.html
        -->
        <managementContext>
            <managementContext createConnector="false"/>
        </managementContext>
 
        <!--
            Configure message persistence for the broker. The default persistence
            mechanism is the KahaDB store (identified by the kahaDB tag).
            For more information, see:
            http://activemq.apache.org/persistence.html
        
        <persistenceAdapter>
            <kahaDB directory="${activemq.data}/kahadb"/>
        </persistenceAdapter>
		-->
		 
         <persistenceAdapter>
            <jdbcPersistenceAdapter dataDirectory="${activemq.data}/activemq-data" dataSource="#mysql-ds"/>
		</persistenceAdapter>
   
		<transportConnectors>
			<transportConnector name="default" uri="tcp://localhost:61616"/>
		</transportConnectors>
 
          <!--
            The systemUsage controls the maximum amount of space the broker will
            use before disabling caching and/or slowing down producers. For more information, see:
            http://activemq.apache.org/producer-flow-control.html
          -->
          <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage percentOfJvmHeap="70" />
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="100 gb"/>
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="50 gb"/>
                </tempUsage>
            </systemUsage>
        </systemUsage>
 
        <!--
            The transport connectors expose ActiveMQ over a given protocol to
            clients and other brokers. For more information, see:
            http://activemq.apache.org/configuring-transports.html
        -->
		 <!--
        <transportConnectors>
            DOS protection, limit concurrent connections to 1000 and frame size to 100MB 
            <transportConnector name="openwire" uri="tcp://127.0.0.1:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="amqp" uri="amqp://127.0.0.1:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="stomp" uri="stomp://127.0.0.1:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="mqtt" uri="mqtt://127.0.0.1:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="ws" uri="ws://127.0.0.1:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
        </transportConnectors>
		
		-->
 
        <!-- destroy the spring context on shutdown to stop jetty -->
        <shutdownHooks>
            <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
        </shutdownHooks>
 
    </broker>
 
    <!--
        Enable web consoles, REST and Ajax APIs and demos
        The web consoles requires by default login, you can disable this in the jetty.xml file
        Take a look at ${ACTIVEMQ_HOME}/conf/jetty.xml for more details
    -->
    <import resource="jetty.xml"/>
 
</beans>
<!-- END SNIPPET: example -->