Spring JMS和ActiveMQ的应用

    笔者近期参与一个分析log的项目。主要流程是：读取Log文件，对每一行Text解析成对应的Object，解析器会将多个Object存放到一个List中并发送到ActiveMQ的Queue中，即Queue中的一个Message即应一个Objects List。然后数据处理thread会consume存放在Queue中的Message，并将处理的结果保存到db。

 

    采用JMS来实现读取Log和分析Log之间的异步运行，使用ActiveMQ的可持久化的Queue，当Message被放进Queue中并持久化后，就会更新Log的读取进度，这样即使程序break down，也不会导致数据被漏掉。

 

    使用Spring来配置JMS，则可以简化和方便JMS的使用，同时还可以使用到Transaction Management。

 

    由于程序并不是很复杂，同时也不需要单独提供ActiveMQ server来运行，所以这里使用的是embed的方式。

 

    总的来说，程序运行起来之后，将会启动以下几个Service：

 

    1. JMS service，这个service会启动一个embed broker。

    2. Data Reader & Parser service，这个service会引用一个JMS Message Provider，用于发送Message到Queue中。

    3. 使用Message Listener Containers来监听Queue，并使用MDP(Message Drive POJO)的方法，来处理并消费掉Message。

 

    本文主要说明如何通过Spring的配置来实现JMS和ActiveMQ的应用，对于这个程序的其他的代码不涉及。

 

    第一步，JMS Service并start embed broker

public class JMSService extends AbstractService {

    private BrokerService broker;

    private String mqConfigFile = "xbean:activemq.xml";
    
    public void start() throws Exception {
        if(broker == null) {            
            broker = BrokerFactory.createBroker(new URI(mqConfigFile));
            broker.start();
            broker.waitUntilStarted();
        }
        super.start();
    }

    public void stop() throws Exception {
        if(broker != null && broker.isStarted()) {
            broker.stop();
            broker.waitUntilStopped();
        }
        super.stop();
    }

}

 

    这里使用BrokerService broker = BrokerFactory.createBroker(new URI(someURI));来创建一个broker，关于someURI的配置，详见这里：http://activemq.apache.org/broker-configuration-uri.html

    我这里的xbean: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.
-->
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:amq="http://activemq.apache.org/schema/core" 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-2.0.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>mw.properties</value>
        </property>
    </bean>

    <!-- The <broker> element is used to configure the ActiveMQ broker. -->
    <broker useJmx="true" xmlns="http://activemq.apache.org/schema/core"
        brokerName="${activemq.broker.name}" dataDirectory="${activemq.broker.datadir}">
        <destinationPolicy>
            <policyMap>
                <policyEntries>
                    <policyEntry topic=">" producerFlowControl="true"
                        memoryLimit="10mb">
                        <pendingSubscriberPolicy>
                            <vmCursor />
                        </pendingSubscriberPolicy>
                    </policyEntry>
                    <policyEntry queue="mdp.queue." producerFlowControl="true"
                        memoryLimit="128mb">
                    </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="true" />
        </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.broker.datadir}/kahadb"
                journalMaxFileLength="96mb" indexWriteBatchSize="10000"
                indexCacheSize="100000" />
        </persistenceAdapter>

        <plugins>
            <!-- Configure authentication; Username, passwords and groups -->
            <simpleAuthenticationPlugin>
                <users>
                    <authenticationUser username="${activemq.userName}"
                        password="${activemq.password}" groups="users,admins" />
                    <authenticationUser username="admin" password="pass"
                        groups="users" />
                    <authenticationUser username="guest" password="pass"
                        groups="guests" />
                </users>
            </simpleAuthenticationPlugin>


            <!-- Lets configure a destination based authorization mechanism -->
            <authorizationPlugin>
                <map>
                    <authorizationMap>
                        <authorizationEntries>
                            <authorizationEntry queue=">" read="admins"
                                write="admins" admin="admins" />
                            <authorizationEntry queue="USERS.>" read="users"
                                write="users" admin="users" />
                            <authorizationEntry queue="GUEST.>" read="guests"
                                write="guests,users" admin="guests,users" />

                            <authorizationEntry queue="TEST.Q" read="guests"
                                write="guests" />

                            <authorizationEntry topic=">" read="admins"
                                write="admins" admin="admins" />
                            <authorizationEntry topic="USERS.>" read="users"
                                write="users" admin="users" />
                            <authorizationEntry topic="GUEST.>" read="guests"
                                write="guests,users" admin="guests,users" />

                            <authorizationEntry topic="ActiveMQ.Advisory.>"
                                read="guests,users" write="guests,users" admin="guests,users" />
                        </authorizationEntries>
                    </authorizationMap>
                </map>
            </authorizationPlugin>
        </plugins>

        <!-- The systemUsage controls the maximum amount of space the broker will 
            use before slowing down producers. For more information, see: http://activemq.apache.org/producer-flow-control.html -->
        <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage limit="2048 mb" />
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="32 gb" name="mdp" />
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="1024 mb" />
                </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>
            <transportConnector name="openwire" uri="${activemq.broker.url}" />
        </transportConnectors>

    </broker>

    <!-- Uncomment to enable Camel Take a look at activemq-camel.xml for more 
        details <import resource="camel.xml"/> -->

    <!-- Enable web consoles, REST and Ajax APIs and demos Take a look at activemq-jetty.xml 
        for more details <import resource="jetty.xml"/> -->
</beans>

 

    第二步，配置JMS ConnectionFactory和Destination

	<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<property name="brokerURL">
			<value>vm://localhost:61616</value>
		</property>
	</bean>

	<!-- Queue -->
	<bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg index="0">
			<value>UserMessageQueue</value>
		</constructor-arg>
	</bean>

 

    第三步，配置JMS Template，以及Message converter

	<!-- Spring JmsTemplate config -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory">
			<!-- lets wrap in a pool to avoid creating a connection per send -->
			<bean class="org.springframework.jms.connection.SingleConnectionFactory">
				<property name="targetConnectionFactory" ref="jmsConnectionFactory" />
			</bean>
		</property>
		<!-- custom MessageConverter -->
		<!-- no message converter configured, so it'll use the SimpleMessageConverter  -->
	</bean>

    注意，这里使用了一个 SingleConnectionFactory，这是spring对ConnectionFactory的一个实现，这个实现会在调用createConnection的地方返回相同的Connection并且忽略所有的close()方法，这样多个JMSTemplate可以共用一个相同的connection，避免每次都重复创建connection造成资源的浪费。

    并且，我没有配置MessageConverter，而且使用Spring自带的SimpleMessageConverter，这是默认选项，不配就是使用SimpleMessageConverter。

 

    第四步，配置JMS Message producer

	<!-- POJO which send Message uses Spring JmsTemplate -->
	<bean id="userMessageProducer" class="cn.lettoo.meetingwatch.jms.UserMessageProducer">
		<property name="template" ref="jmsTemplate" />
		<property name="destination" ref="destinationQueue" />
	</bean>

    Message Producer通过JMS Template和destination来发送Message。下面是代码实现：

public class UserMessageProducer implements IMessageProducer {

    protected JmsTemplate template; 

    protected Queue destination; 

    public void setTemplate(JmsTemplate template) { 
       this.template = template; 
    } 

    public void setDestination(Queue destination) { 
       this.destination = destination; 
    }

    public void send(Object object) {
        template.convertAndSend(this.destination, object); 
    } 

}

 因为我并没有为JMS template指定特定的MessageConverter，所以这里template.convertAndSend()应该就是SimpleMessageConverter.convertAndSend()来实现的。

 

    第五步，配置JMS Message Consumer

    我并没有在spring的配置文件单独配置一个Message Consumer，这是一个非常简单的POJO，这里只是为测试写的一个简单示例：只是简单把读到的Message打印出来。

public class UserMessageConsumer {

    public void printUser(User user) {
        user.getId();
        user.getName();
        System.out.println(user);
    }

}

 

    第六步，配置JMS Message Listener Container以及Listerer

	<!-- Message Driven POJO -->
	<bean id="messageListener"
		class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<constructor-arg>
			<bean class="cn.lettoo.meetingwatch.jms.UserMessageConsumer">
			</bean>
		</constructor-arg>
		<property name="defaultListenerMethod" value="printUser" />
	</bean>

	<!-- listener container -->
	<bean id="listenerContainer"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="jmsConnectionFactory" />
		<property name="destination" ref="destinationQueue" />
		<property name="messageListener" ref="messageListener" />
	</bean>

    这里使用org.springframework.jms.listener.adapter.MessageListenerAdapter来实现一个Listener，这样做的好处是，可以使用非常简单的POJO来作为一个consumer，就是上面的UserMessageConsumer一样。当Queue里有一条Message的时候，会使用UserMessageConsumer的printUser来消费掉这条Message。

    Spring JMS提供3种ListenerContainer，最常使用的是DefaultMessageListenerContainer。这里把Listener注入，来实现对Message的Listener。

 

参考：

1. Chapter 19. JMS (Java Message Service)

2. ActiveMQ5.0实战三:使用Spring发送,消费topic和queue消息