2. 程式人生 > >ActiveMQ將訊息持久化到資料庫

ActiveMQ將訊息持久化到資料庫

阿新 • • 發佈：2019-01-01
需求描述：將ActiveMQ訊息持久化到mySql\Oracle資料庫中 ；

環境描述：目前最新版本是ActiveMQ5.13.2，本文講述的例項是ActiveMQ5.9.0。

使用預設的持久化機制，我們不容易直接看到訊息究竟是如何持久的。ActiveMQ提供的JDBC持久化機制，能夠將持久化資訊儲存到資料庫。通過檢視資料庫中ActiveMQ生成的表結構和儲存的資料，能夠幫助我們更好的瞭解訊息的持久化機制。現在介紹如何配置activemq，將資料持久化到mysql中。

1.配置activeMQ需要的mySql資料來源

為了能夠使用JDBC訪問mysql資料庫，顯然必須要配置訊息伺服器的資料庫源。在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>
在</broker>結點之後，增加資料來源的配置

2.改變activeMQ預設的持久化方式

在activemq.xml中註釋掉預設的kahadb，使用jdbc持久化

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

3.設定JMS訪問連線協議型別

修改原來的協議連線為如下方式：

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

4.提供mysql的驅動程式

由於activeMQ訊息伺服器，沒有自帶mysql資料庫的驅動程式。我們需要手動將mysql驅動新增到訊息伺服器。

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

經過上面的三步配置，我們重新啟動訊息伺服器，就可以發現activeMQ在資料庫中新建了3張表activemq_acks  ，activemq_lock  ，activemq_msgs 。


至此資料庫持久化完成。ActiveMQ持久化的中表結構是什麼，表需要人工建立嗎？其實不需要，ActiveMQ會幫助我們生成的。只需要制定採用的資料庫名稱並，建立資料庫即可。以為為ActiveMQ採用MySQL5.7持久化產生的SQL語句：

持久化mysql資料庫的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 -->

6.參考資料

相關推薦

ActiveMQ將訊息持久化到資料庫

需求描述：將ActiveMQ訊息持久化到mySql\Oracle資料庫中 ；環境描述：目前最新版本是ActiveMQ5.13.2，本文講述的例項是ActiveMQ5.9.0。 使用預設的持久化機制，我們不容易直接看到訊息究竟是如何持久的。ActiveMQ提供的JDBC持久化

ActiveMQ深入淺出(五)——ActiveMQ將訊息持久化到資料庫

需求描述：將ActiveMQ訊息持久化到mySql\Oracle資料庫中 ； 環境描述：目前最新版本是ActiveMQ5.13.2，本文講述的例項是ActiveMQ5.9.0。 使用預設的持久化機制，我們不容易直接看到訊息究竟是如何持久的。ActiveMQ提供

JMS學習九 ActiveMQ的訊息持久化到Mysql資料庫

1、將連線Mysql資料庫驅動包，放到ActiveMQ的lib目錄下 2，修改ActiveMQ的conf目錄下的active.xml檔案，修改資料持久化的方式          2.1  修改原來的kshadb的持久化資料

JMS訊息持久化，將ActiveMQ訊息持久化到mySql資料庫中

ActiveMQ5.8.0版本採用kahadb作為預設的訊息持久化方式。使用預設的持久化機制，我們不容易直接看到訊息究竟是如何持久的。ActiveMQ提供的JDBC持久化機制，能夠將持久化資訊儲存到資

activemq 學習系列(三) 訊息持久化到MySql資料庫

ActiveMq 訊息持久化到MySql資料庫 1、修改 conf/activemq.xml配置檔案 <persistenceAdapter> <kahaDB directory="${activemq.data}/kahadb"/> </persistence

ActiveMQ開啟使用者認證及訊息持久化到資料庫功能

1、開啟使用者認證機制 ActiveMQ預設是不開啟使用者認證機制的，如果需要開啟則需要修改配置。 1. 首先開啟activemq.xml，在broker下增加 <plugins> <jaasAuthenticationPlu

ActiveMQ 訊息持久化到資料庫

1：前言      這一段給公司開發訊息匯流排有機會研究ActiveMQ，今天撰文給大家介紹一下他的持久化訊息。本文只介紹三種方式,分別是持久化為檔案，MYSql，Oracle。下面逐一介紹。 A：持久化為檔案      這個你裝ActiveMQ時預設就是這種，只要

ActiveMQ學習筆記（8）----ActiveMQ的訊息儲存持久化

1. 概述 　　ActiveMQ不僅支援persistent和non-persistent兩種方式，還支援訊息的恢復（recovery）方式。 2. PTP 　　Queue的儲存是很簡單的，其實就是FIFO的Queue 　　　　 2. PUB/SUB 　　對於持久化訂閱主題，每一個消費者都將獲得一

ActiveMQ發訊息、收訊息、持久化，查詢佇列剩餘訊息數、出隊數的實現

1.首先啟動activeMQ的服務 public class RunServer {           /** 啟動activeMQ服務&

activemq實戰之訊息持久化

對於activemq訊息的持久化我們在第二節的時候就簡單介紹過，今天我們詳細的來分析一下activemq的持久化過程以及持久化外掛。在生產環境中為確保訊息的可靠性，我們肯定的面臨持久化訊息的問題，今天就一起來攻克他吧。 1. 持久化方式介紹 前面我們也簡單提

ActiveMQ中訊息的重發與持久化儲存

訊息中介軟體解決方案續 　　上一篇中我們講到了在Spring工程中基本的使用訊息中介軟體，這裡就不在繼續贅述。 　　針對訊息中介軟體，這篇講解兩個我們常遇到的問題。 　　問題1：如果我們的訊息的接收過程中發生異常，怎麼解決？ 　　問題2：釋出訂閱模式（Topic）下如果消費端宕機引起的訊息的丟失，怎麼

ActiveMQ的幾種訊息持久化機制

為了避免意外宕機以後丟失資訊，需要做到重啟後可以恢復訊息佇列，訊息系統一般都會採用持久化機制。 ActiveMQ的訊息持久化機制有JDBC，AMQ，KahaDB和LevelDB四種方式，無論使用哪種持久化方式，訊息的儲存邏輯都是一致的。 就是在傳送者將訊息傳送出

ActiveMQ 訊息持久化

生產端（訊息持久化）： import javax.jms.Connection; import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.MessageProducer; import java

ActiveMQ的JDBC訊息持久化機制

為了避免意外宕機以後丟失資訊，需要做到重啟後可以恢復訊息佇列，訊息系統一般都會採用持久化機制。 ActiveMQ的訊息持久化機制有JDBC，AMQ，KahaDB和LevelDB，無論使用哪種持久化方式，訊息的儲存邏輯都是一致的。 就是在傳送者將訊息傳送出去後，訊息中心

ActiveMQ訊息持久化儲存策略

   ActiveMQ的核心是Java編寫的，也就是說如果服務端沒有Java執行環境ActiveMQ是無法執行的。ActiveMQ啟動時，啟動指令碼使用wrapper包裝器來啟動JVM。JVM相關的配置資訊在啟動目錄的“wrapper.conf”配置檔案中。各位讀者可以通過改變其中的配置項

循序漸進ActiveMQ（3）----MessageConsumer的訊息選擇器及mysql訊息持久化

MessageConsumer的訊息選擇器MessageConsumer是一個由Session建立的物件，用來從Destination接收訊息。看一下Session建立MessageConsumer的構造方法有哪些： public MessageConsumer create

Activemq訊息持久化

<bean id="ds-mysql" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close"> <property name="driverClassName" value="com.mysql

JMS之activemq訊息持久化

JMS 即 java message service 是為java提供了一種建立、傳送、接收訊息的通用方法。可以將複雜的系統進行業務分離，變成靈活的高度解耦合的佈局。同時對我們的日常業務需求開發，提供了非常靈活的業務解決方案。比如繳費還款送積分，送積分的業務邏輯不能影響到繳

ActiveMQ使用筆記（二）ActiveMQ訊息持久化一

<persistenceAdapter>   <mKahaDB directory="${activemq.base}/data/kahadb">     <filteredPersistenceAdapters>       <!-- match all queu

訊息中介軟體-activemq實戰之訊息持久化(六)

對於activemq訊息的持久化我們在第二節的時候就簡單介紹過，今天我們詳細的來分析一下activemq的持久化過程以及持久化外掛。在生產環境中為確保訊息的可靠性，我們肯定的面臨持久化訊息的問題，今天就一起來攻克他吧。 1. 持久化方式介紹 前面我們也簡