spring-jms

这里写篇文章记录一下jms持久化到mysqlde  使用方法。这里先介绍一下场景，

     ActiveMQ5.9.0    http://activemq.apache.org/

     开发环境搭建使用的是spring-jms

下面讲一下持久化方法：按照网上相关资料搜索查明，需要修改<ActiveMQ_dir>/conf/activemq.xml里的持久化节点：

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

 增加mysql数据源配置如下：

 

 

<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=""/>
		<property name="maxActive" value="200"/>
		<property name="poolPreparedStatements" value="true"/>
</bean>

 这里写一下需要注意的事情：

 

1. activemq数据库的编码集latin1，UTF-8 会创建不成功
2. 拷贝mysql数据库连接包到<ActiveMQ_dir>/lib包下                             
3.         对出现异常但是没有处理的消息我们如何重复利用。这里我说一下我的简单方法，其他方法还有待发现。访问程序http://localhost:8161/admin/queues.jsp，用户名admin,密码admin .操作重新发送。具体如图。
4. 
   
    

更多学习指导可以查看那

<ActiveMQ_dir>\docs\user-guide.html

需要将无效的消息置活，所以需要设置失败的消息，如下：修改<ActiveMQ_dir>/conf/activemq.xml

 

<policyEntry queue=">">
					<deadLetterStrategy>
						<individualDeadLetterStrategy
								queuePrefix="DLQ." useQueueForQueueMessages="true" processExpired="true"
								processNonPersistent="true" />
					</deadLetterStrategy>
				</policyEntry>

 x

修改后的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>

    <!--
        The <broker> element is used to configure the ActiveMQ broker.
    -->
    <broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost"  persistent="true"  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>
				 
				<policyEntry queue=">">
					<deadLetterStrategy>
						<individualDeadLetterStrategy
								queuePrefix="DLQ." useQueueForQueueMessages="true" processExpired="true"
								processNonPersistent="true" />
					</deadLetterStrategy>
				</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"/>-->
			<jdbcPersistenceAdapter dataSource="#mysql-ds"/>
        </persistenceAdapter>
		 
 

          <!--
            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://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;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"/>


	<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=""/>
		<property name="maxActive" value="200"/>
		<property name="poolPreparedStatements" value="true"/>
	</bean>

</beans>
<!-- END SNIPPET: example -->

 

下面是

读取并处理deadLetter的方法

Message message = jmsTemplate.receive(queueName);//例如：DLQ.myqueue

浏览deadLetterne内消息的方法

Integer result = jmsTemplate.browse(queueName , new BrowserCallback<Integer>() {
			public Integer doInJms(Session session, QueueBrowser browser) throws JMSException {
				Enumeration<?> messages = browser.getEnumeration();
				int total = 0;
				while (messages.hasMoreElements()) {
					Message message = (Message) messages.nextElement();
					if (message instanceof ObjectMessage) { 
						ObjectMessage objMessage = (ObjectMessage) message;
						Object obj = objMessage.getObject();
						 SerializableEntity entity = (SerializableEntity) obj;  
						 System.out.println("接收到一个ObjectMessage，包含对象为："+entity+",对象内容是："+entity.getMsg());
					
					}
					total++;
				}
				return total;
			}
		});