Spring集成ActiveMQ配置

1.       集成环境

Spring采用2.5.6版本,ActiveMQ使用的是5.4.2,从apache站点可以下载。本文是将Spring集成ActiveMQ来发送和接收JMS消息。

2.       集成步骤

将下载的ActiveMQ解压缩后文件夹如下
Spring集成ActiveMQ配置
 activemq-all-5.4.2.jar是activemq的所有的类jar包。lib下面是模块分解后的jar包。将lib下面的

/lib/activation-1.1.jar   

/lib/activemq-camel-5.4.2.jar   


/lib/activemq-console-5.4.2.jar   


/lib/activemq-core-5.4.2.jar   


/lib/activemq-jaas-5.4.2.jar   


/lib/activemq-pool-5.4.2.jar   


/lib/activemq-protobuf-1.1.jar   


/lib/activemq-spring-5.4.2.jar   


/lib/activemq-web-5.4.2.jar  
/lib/activation-1.1.jar  

/lib/activemq-camel-5.4.2.jar  


/lib/activemq-console-5.4.2.jar  


/lib/activemq-core-5.4.2.jar  


/lib/activemq-jaas-5.4.2.jar  


/lib/activemq-pool-5.4.2.jar  


/lib/activemq-protobuf-1.1.jar  


/lib/activemq-spring-5.4.2.jar  


/lib/activemq-web-5.4.2.jar  
/lib/activation-1.1.jar
/lib/activemq-camel-5.4.2.jar
/lib/activemq-console-5.4.2.jar
/lib/activemq-core-5.4.2.jar
/lib/activemq-jaas-5.4.2.jar
/lib/activemq-pool-5.4.2.jar
/lib/activemq-protobuf-1.1.jar
/lib/activemq-spring-5.4.2.jar
/lib/activemq-web-5.4.2.jar

文件全部拷贝到项目中。

而Spring项目所需要的jar包如下 

/lib/spring-beans-2.5.6.jar   

/lib/spring-context-2.5.6.jar   


/lib/spring-context-support-2.5.6.jar   


/lib/spring-core-2.5.6.jar   


/lib/spring-jms-2.5.6.jar   

/lib/spring-tx.jar  
/lib/spring-beans-2.5.6.jar  

/lib/spring-context-2.5.6.jar  


/lib/spring-context-support-2.5.6.jar  


/lib/spring-core-2.5.6.jar  


/lib/spring-jms-2.5.6.jar  

/lib/spring-tx.jar  
/lib/spring-beans-2.5.6.jar
/lib/spring-context-2.5.6.jar
/lib/spring-context-support-2.5.6.jar
/lib/spring-core-2.5.6.jar
/lib/spring-jms-2.5.6.jar
/lib/spring-tx.jar

当然还需要一些其他的jar文件

/lib/geronimo-j2ee-management_1.1_spec-1.0.1.jar   

/lib/jms-1.1.jar   


/lib/log4j-1.2.15.jar   


/lib/slf4j-api-1.6.1.jar   


/lib/slf4j-nop-1.6.1.jar  
/lib/geronimo-j2ee-management_1.1_spec-1.0.1.jar  

/lib/jms-1.1.jar  


/lib/log4j-1.2.15.jar  


/lib/slf4j-api-1.6.1.jar  


/lib/slf4j-nop-1.6.1.jar  
/lib/geronimo-j2ee-management_1.1_spec-1.0.1.jar
/lib/jms-1.1.jar
/lib/log4j-1.2.15.jar
/lib/slf4j-api-1.6.1.jar
/lib/slf4j-nop-1.6.1.jar

项目的依赖jar都准备好后就可以写配置文件了。

 Spring配置文件

配置文件内容如下

<?xml version="1.0" encoding="UTF-8"?>   

<beans xmlns="http://www.springframework.org/schema/beans"  


    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"  


    xsi:schemaLocation="http://www.springframework.org/schema/beans    


        http://www.springframework.org/schema/beans/spring-beans-2.5.xsd    


        http://www.springframework.org/schema/context    


        http://www.springframework.org/schema/context/spring-context-2.5.xsd"   


    default-autowire="byName">   

  
  
    <!-- 配置connectionFactory -->   

    <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory"  


        destroy-method="stop">   


        <property name="connectionFactory">   


            <bean class="org.apache.activemq.ActiveMQConnectionFactory">   


                <property name="brokerURL">   


                    <value>tcp://127.0.0.1:61616</value>   

                </property>   
            </bean>   
        </property>   

        <property name="maxConnections" value="100"></property>   

    </bean>   
  
    <!-- Spring JMS Template -->   

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">   


        <property name="connectionFactory">   


            <ref local="jmsFactory" />   

        </property>   

        <property name="defaultDestinationName" value="subject" />   


        <!-- 区别它采用的模式为false是p2p为true是订阅 -->   


        <property name="pubSubDomain" value="true" />   

    </bean>   
  
    <!-- 发送消息的目的地(一个队列) -->   

    <bean id="destination" class="org.apache.activemq.command.ActiveMQTopic">   

        <!-- 设置消息队列的名字 -->   

        <constructor-arg index="0" value="subject" />   

    </bean>   
  
    <!-- 消息监听     -->   

    <bean id="listenerContainer"  


        class="org.springframework.jms.listener.DefaultMessageListenerContainer">   


        <property name="concurrentConsumers" value="10" />   


        <property name="connectionFactory" ref="jmsFactory" />   


        <property name="destinationName" value="subject" />   


        <property name="messageListener" ref="messageReceiver" />   


        <property name="pubSubNoLocal" value="false"></property>   

    </bean>   
  

    <bean id="messageReceiver"  


        class="com.liuyan.jms.consumer.ProxyJMSConsumer">   


        <property name="jmsTemplate" ref="jmsTemplate"></property>   

    </bean>   
</beans>  
<?xml version="1.0" encoding="UTF-8"?>  

<beans xmlns="http://www.springframework.org/schema/beans"  


    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"  


    xsi:schemaLocation="http://www.springframework.org/schema/beans    


        http://www.springframework.org/schema/beans/spring-beans-2.5.xsd    


        http://www.springframework.org/schema/context    


        http://www.springframework.org/schema/context/spring-context-2.5.xsd"   


    default-autowire="byName">  

  
  
    <!-- 配置connectionFactory -->  

    <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory"  


        destroy-method="stop">  


        <property name="connectionFactory">  


            <bean class="org.apache.activemq.ActiveMQConnectionFactory">  


                <property name="brokerURL">  


                    <value>tcp://127.0.0.1:61616</value>   

                </property>  
            </bean>  
        </property>  

        <property name="maxConnections" value="100"></property>  

    </bean>  
  
    <!-- Spring JMS Template -->  

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  


        <property name="connectionFactory">  


            <ref local="jmsFactory" />  

        </property>  

        <property name="defaultDestinationName" value="subject" />  


        <!-- 区别它采用的模式为false是p2p为true是订阅 -->  


        <property name="pubSubDomain" value="true" />  

    </bean>  
  
    <!-- 发送消息的目的地(一个队列) -->  

    <bean id="destination" class="org.apache.activemq.command.ActiveMQTopic">  

        <!-- 设置消息队列的名字 -->  

        <constructor-arg index="0" value="subject" />  

    </bean>  
  
    <!-- 消息监听     -->  

    <bean id="listenerContainer"  


        class="org.springframework.jms.listener.DefaultMessageListenerContainer">  


        <property name="concurrentConsumers" value="10" />  


        <property name="connectionFactory" ref="jmsFactory" />  


        <property name="destinationName" value="subject" />  


        <property name="messageListener" ref="messageReceiver" />  


        <property name="pubSubNoLocal" value="false"></property>  

    </bean>  
  

    <bean id="messageReceiver"  


        class="com.liuyan.jms.consumer.ProxyJMSConsumer">  


        <property name="jmsTemplate" ref="jmsTemplate"></property>  

    </bean>  
</beans>  
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
	xsi:schemaLocation="http://www.springframework.org/schema/beans 
        http://www.springframework.org/schema/beans/spring-beans-2.5.xsd 
        http://www.springframework.org/schema/context 
        http://www.springframework.org/schema/context/spring-context-2.5.xsd"
	default-autowire="byName">


	<!-- 配置connectionFactory -->
	<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
		destroy-method="stop">
		<property name="connectionFactory">
			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
				<property name="brokerURL">
					<value>tcp://127.0.0.1:61616</value>
				</property>
			</bean>
		</property>
		<property name="maxConnections" value="100"></property>
	</bean>

	<!-- Spring JMS Template -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory">
			<ref local="jmsFactory" />
		</property>
		<property name="defaultDestinationName" value="subject" />
		<!-- 区别它采用的模式为false是p2p为true是订阅 -->
		<property name="pubSubDomain" value="true" />
	</bean>

	<!-- 发送消息的目的地(一个队列) -->
	<bean id="destination" class="org.apache.activemq.command.ActiveMQTopic">
		<!-- 设置消息队列的名字 -->
		<constructor-arg index="0" value="subject" />
	</bean>

	<!-- 消息监听     -->
	<bean id="listenerContainer"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="concurrentConsumers" value="10" />
		<property name="connectionFactory" ref="jmsFactory" />
		<property name="destinationName" value="subject" />
		<property name="messageListener" ref="messageReceiver" />
		<property name="pubSubNoLocal" value="false"></property>
	</bean>

	<bean id="messageReceiver"
		class="com.liuyan.jms.consumer.ProxyJMSConsumer">
		<property name="jmsTemplate" ref="jmsTemplate"></property>
	</bean>
</beans>

编写代码

消息发送者:这里面消息生产者并没有在Spring配置文件中进行配置,这里仅仅使用了Spring中的JMS模板和消息目的而已。 

public class HelloSender {   
  

    /**  
     * @param args  
     */  

    public static void main(String[] args) {   


        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(   


                new String[] { "classpath:/spring/applicationContext-jms.xml" });   

           
        JmsTemplate template = (JmsTemplate) applicationContext   

                .getBean("jmsTemplate");   

           
        Destination destination = (Destination) applicationContext   

                .getBean("destination");   

  

        template.send(destination, new MessageCreator() {   


            public Message createMessage(Session session) throws JMSException {   


                return session   


                        .createTextMessage("发送消息:Hello ActiveMQ Text Message!");   

            }   
        });   

        System.out.println("成功发送了一条JMS消息");   

  
    }   
  
}  
public class HelloSender {  
  

    /** 
     * @param args 
     */  

    public static void main(String[] args) {  


        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(  


                new String[] { "classpath:/spring/applicationContext-jms.xml" });  

          
        JmsTemplate template = (JmsTemplate) applicationContext  

                .getBean("jmsTemplate");  

          
        Destination destination = (Destination) applicationContext  

                .getBean("destination");  

  

        template.send(destination, new MessageCreator() {  


            public Message createMessage(Session session) throws JMSException {  


                return session  


                        .createTextMessage("发送消息:Hello ActiveMQ Text Message!");  

            }  
        });  

        System.out.println("成功发送了一条JMS消息");  

  
    }  
  
}  
public class HelloSender {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		ApplicationContext applicationContext = new ClassPathXmlApplicationContext(
				new String[] { "classpath:/spring/applicationContext-jms.xml" });
		
		JmsTemplate template = (JmsTemplate) applicationContext
				.getBean("jmsTemplate");
		
		Destination destination = (Destination) applicationContext
				.getBean("destination");

		template.send(destination, new MessageCreator() {
			public Message createMessage(Session session) throws JMSException {
				return session
						.createTextMessage("发送消息:Hello ActiveMQ Text Message!");
			}
		});
		System.out.println("成功发送了一条JMS消息");

	}

}

消息接收

/**  
 * JMS消费者  
 *   
 *   
 * <p>  
 * 消息题的内容定义  
 * <p>  
 * 消息对象 接收消息对象后: 接收到的消息体* <p>   
 */  

public class ProxyJMSConsumer {   

  

    public ProxyJMSConsumer() {   

  
    }   
  

    private JmsTemplate jmsTemplate;   

  

    public JmsTemplate getJmsTemplate() {   


        return jmsTemplate;   

    }   
  

    public void setJmsTemplate(JmsTemplate jmsTemplate) {   


        this.jmsTemplate = jmsTemplate;   

    }   
  

    /**  
     * 监听到消息目的有消息后自动调用onMessage(Message message)方法  
     */  

    public void recive() {   


        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(   


                new String[] { "classpath:/spring/applicationContext-jms.xml" });   

  
        Destination destination = (Destination) applicationContext   

                .getBean("destination");   

  

        while (true) {   

  

            try {   

                TextMessage txtmsg = (TextMessage) jmsTemplate   
                        .receive(destination);   

                if (null != txtmsg) {   


                    System.out.println("[DB Proxy] " + txtmsg);   


                    System.out.println("[DB Proxy] 收到消息内容为: "  

                            + txtmsg.getText());   

                } else  


                    break;   


            } catch (Exception e) {   

                e.printStackTrace();   
            }   
  
        }   
    }   
  
}  
/** 
 * JMS消费者 
 *  
 *  
 * <p> 
 * 消息题的内容定义 
 * <p> 
 * 消息对象 接收消息对象后: 接收到的消息体* <p>  
 */  

public class ProxyJMSConsumer {  

  

    public ProxyJMSConsumer() {  

  
    }  
  

    private JmsTemplate jmsTemplate;  

  

    public JmsTemplate getJmsTemplate() {  


        return jmsTemplate;  

    }  
  

    public void setJmsTemplate(JmsTemplate jmsTemplate) {  


        this.jmsTemplate = jmsTemplate;  

    }  
  

    /** 
     * 监听到消息目的有消息后自动调用onMessage(Message message)方法 
     */  

    public void recive() {  


        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(  


                new String[] { "classpath:/spring/applicationContext-jms.xml" });  

  
        Destination destination = (Destination) applicationContext  

                .getBean("destination");  

  

        while (true) {  

  

            try {  

                TextMessage txtmsg = (TextMessage) jmsTemplate  
                        .receive(destination);  

                if (null != txtmsg) {  


                    System.out.println("[DB Proxy] " + txtmsg);  


                    System.out.println("[DB Proxy] 收到消息内容为: "  

                            + txtmsg.getText());  

                } else  


                    break;  


            } catch (Exception e) {  

                e.printStackTrace();  
            }  
  
        }  
    }  
  
}  
/**
 * JMS消费者
 * 
 * 
 * <p>
 * 消息题的内容定义
 * <p>
 * 消息对象 接收消息对象后: 接收到的消息体* <p> 
 */
public class ProxyJMSConsumer {

	public ProxyJMSConsumer() {

	}

	private JmsTemplate jmsTemplate;

	public JmsTemplate getJmsTemplate() {
		return jmsTemplate;
	}

	public void setJmsTemplate(JmsTemplate jmsTemplate) {
		this.jmsTemplate = jmsTemplate;
	}

	/**
	 * 监听到消息目的有消息后自动调用onMessage(Message message)方法
	 */
	public void recive() {
		ApplicationContext applicationContext = new ClassPathXmlApplicationContext(
				new String[] { "classpath:/spring/applicationContext-jms.xml" });

		Destination destination = (Destination) applicationContext
				.getBean("destination");

		while (true) {

			try {
				TextMessage txtmsg = (TextMessage) jmsTemplate
						.receive(destination);
				if (null != txtmsg) {
					System.out.println("[DB Proxy] " + txtmsg);
					System.out.println("[DB Proxy] 收到消息内容为: "
							+ txtmsg.getText());
				} else
					break;
			} catch (Exception e) {
				e.printStackTrace();
			}

		}
	}

}

这里边也是并不是直接使用Spring来初始化建立消息消费者实例,而是在此消费者注入了JMS模板而已。

写一个main入口,初始化消息消费者  

public class JMSTest {   
  

    /**  
     * @param args  
     */  

    public static void main(String[] args) {   


        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(   


                new String[] { "classpath:/spring/applicationContext-jms.xml" });   

           
        ProxyJMSConsumer proxyJMSConsumer = (ProxyJMSConsumer) applicationContext   

                .getBean("messageReceiver");   

           

        System.out.println("初始化消息消费者");   

    }   
  
}  
public class JMSTest {  
  

    /** 
     * @param args 
     */  

    public static void main(String[] args) {  


        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(  


                new String[] { "classpath:/spring/applicationContext-jms.xml" });  

          
        ProxyJMSConsumer proxyJMSConsumer = (ProxyJMSConsumer) applicationContext  

                .getBean("messageReceiver");  

          

        System.out.println("初始化消息消费者");  

    }  
  
}  
public class JMSTest {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		ApplicationContext applicationContext = new ClassPathXmlApplicationContext(
				new String[] { "classpath:/spring/applicationContext-jms.xml" });
		
		ProxyJMSConsumer proxyJMSConsumer = (ProxyJMSConsumer) applicationContext
				.getBean("messageReceiver");
		
		System.out.println("初始化消息消费者");
	}

}

使用的时候先开启ActiveMQ服务,默认是占用了61616端口。之后开启测试程序,开启2个消息消费者监听。之后再运行消息生产者的代码后,消息就可以被消息消费者接收到了。 

相关推荐