基于Maven的ActiveMQ+spring 多消息队列配置实例

心已赠人 2022-07-15 00:47 51阅读 0赞

# 一、前言： #

本文主要是示范基于Maven的ActiveMQ+spring的多消息队列简单使用，基于ActiveMQ消息代理的Spring JMS消息配置，以springmvc的使用消息队列

# 二、环境： #

在apache官方网站下载后，解压缩安装包，点击apache-activemq-5.9.0\\bin\\activemq.bat运行即可。运行后浏览器输入http://localhost:8161/表明安装成功， 这时就可以使用它进行消息代理了。

# 三、配置部分 #

## 1. maven配置。 ##

引入依赖jar包。pom.xml：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
      
      <modelVersion>4.0.0</modelVersion>
      <groupId>com.worldsconnection</groupId>
      <artifactId>activeMQDemao</artifactId>
      <packaging>war</packaging>
      <version>0.0.1-SNAPSHOT</version>
      <name>gzframe Maven Webapp</name>
      <url>http://maven.apache.org</url>
      
      
      <properties>
          <springframework>4.0.2.RELEASE</springframework>
          <javax.servlet>3.1.0</javax.servlet>
      </properties>
      
      <dependencies>
        
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <version>4.10</version>
          <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>jstl</groupId>
            <artifactId>jstl</artifactId>
            <version>1.2</version>
        </dependency>
        <dependency>    
            <groupId>javax.servlet</groupId>    
            <artifactId>servlet-api</artifactId>
            <version>${javax.servlet}</version>
        </dependency>
        
        
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
            <version>${springframework}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>${springframework}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-tx</artifactId>
            <version>${springframework}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
            <version>${springframework}</version>
        </dependency>
        <dependency>  
            <groupId>org.springframework</groupId>  
            <artifactId>spring-jms</artifactId>  
            <version>${springframework}</version>  
        </dependency>
        
        <dependency>
            <groupId>org.apache.xbean</groupId>
            <artifactId>xbean-spring</artifactId>
            <version>3.16</version>
        </dependency>
        
        
        <dependency>  
            <groupId>org.apache.activemq</groupId>  
            <artifactId>activemq-all</artifactId>  
            <version>5.9.0</version>
        </dependency> 
        <dependency>  
            <groupId>org.apache.activemq</groupId>  
            <artifactId>activemq-pool</artifactId>  
            <version>5.9.0</version>  
        </dependency>  
        
      </dependencies>
      <build>
        <finalName>activeMQDemao</finalName>
      </build>
    </project>

## 2. spring配置（spring-activemq.xml） ##

<?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:amq="http://activemq.apache.org/schema/core"
    	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-5.8.0.xsd">
    
    	
    	<bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    		<constructor-arg index="0" value="tcp://192.168.0.100:61616" />
    		<property name="useAsyncSend" value="true" />
    	</bean>
    
    	
    	<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
    		<constructor-arg ref="amqConnectionFactory" />
    		<property name="sessionCacheSize" value="100" />
    	</bean>
    
    	
    	<bean id="toMSQueue" class="org.apache.activemq.command.ActiveMQQueue">
    		<constructor-arg index="0" value="toMSQueue" />
    	</bean>
    	<bean id="toLDYQueue" class="org.apache.activemq.command.ActiveMQQueue">
    		<constructor-arg index="0" value="toLDYQueue" />
    	</bean>
     
    	
    	<bean id="jmsTemplateToMS" class="org.springframework.jms.core.JmsTemplate">
    		<property name="connectionFactory" ref="connectionFactory" />
    		<property name="defaultDestination" ref="toMSQueue" />
    		<property name="receiveTimeout" value="10000" />
    	</bean>
    	<bean id="jmsTemplateToLDY" class="org.springframework.jms.core.JmsTemplate">
    		<property name="connectionFactory" ref="connectionFactory" />
    		<property name="defaultDestination" ref="toLDYQueue" />
    		<property name="receiveTimeout" value="10000" />
    	</bean>
    
    	
        <bean id="queueMessageListenerToMS" class="com.wanle.ms.mq.QueueMessageListenerToMS" />
        <bean id="queueMessageListenerToLDY" class="com.wanle.ms.mq.QueueMessageListenerToLDY" />
        
        
        <bean id="queueListenerContainerToMS"
            class="org.springframework.jms.listener.DefaultMessageListenerContainer">
            <property name="connectionFactory" ref="connectionFactory" />
            <property name="destination" ref="toMSQueue" />
            <property name="messageListener" ref="queueMessageListenerToMS" />
        </bean>
        <bean id="queueListenerContainerToLDY"
            class="org.springframework.jms.listener.DefaultMessageListenerContainer">
            <property name="connectionFactory" ref="connectionFactory" />
            <property name="destination" ref="toLDYQueue" />
            <property name="messageListener" ref="queueMessageListenerToLDY" />
        </bean>
    </beans>

## 3. 加载springContext.xml  ##

我这里是使用包扫描和注解的.

<?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"
        xmlns:mvc="http://www.springframework.org/schema/mvc"
        xsi:schemaLocation="http://www.springframework.org/schema/beans 
            http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/context
            http://www.springframework.org/schema/context/spring-context-4.0.xsd
            http://www.springframework.org/schema/mvc 
            http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd">
    
        <context:component-scan base-package="com.worldsconnection.mq" />
        <mvc:annotation-driven />
         
    </beans>

## 4. web.xml中加载spring-activemq.xml 和springContext.xml 的配置 ##

<?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"
        xmlns:mvc="http://www.springframework.org/schema/mvc"
        xsi:schemaLocation="http://www.springframework.org/schema/beans 
            http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/context
            http://www.springframework.org/schema/context/spring-context-4.0.xsd
            http://www.springframework.org/schema/mvc 
            http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd">
    
        <context:component-scan base-package="com.worldsconnection.mq" />
        <mvc:annotation-driven />
         
    </beans>

# 四：代码部分： #

## 1.生产者服务（ProducerService.java） ##

package com.wanle.ms.mq;
    
    import javax.annotation.Resource;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Service;
    
    @Service
    public class ProducerService {
    
    	@Resource(name = "jmsTemplateToMS")
    	private JmsTemplate jmsTemplateToMS;
    	
    	@Resource(name = "jmsTemplateToLDY")
    	private JmsTemplate jmsTemplateToLDY;
    
    	/**
    	 * 向指定队列发送消息
    	 */
    	public void sendMessage(Destination destination, final String msg) {
    		System.out.println("向队列" + destination.toString() + "发送了消息------------" + msg);
    		jmsTemplateToMS.send(destination, new MessageCreator() {
    			public Message createMessage(Session session) throws JMSException {
    				return session.createTextMessage(msg);
    			}
    		});
    	}
    
    	/**
    	 * 向默认队列发送消息
    	 */
    	public void sendMessage(final String msg) {
    		String destination = jmsTemplateToMS.getDefaultDestination().toString();
    		System.out.println("向队列" + destination + "发送了消息------------" + msg);
    		jmsTemplateToMS.send(new MessageCreator() {
    			public Message createMessage(Session session) throws JMSException {
    				return session.createTextMessage(msg);
    			}
    		});
    	}
    
    }

## 2.消费者服务（ConsumerService.java） ##

package com.wanle.ms.mq;
    
    import javax.annotation.Resource;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.TextMessage;
    
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.stereotype.Service;
    
    @Service
    public class ConsumerService {
    
    	@Resource(name = "jmsTemplateToMS")
    	private JmsTemplate jmsTemplateToMS;
    	
    	@Resource(name = "jmsTemplateToLDY")
    	private JmsTemplate jmsTemplateToLDY;
    	
    	/**
    	 * 接收消息
    	 */
    	public TextMessage receive(Destination destination) {
    		TextMessage tm = (TextMessage) jmsTemplateToMS.receive(destination);
    		try {
    			System.out.println("从队列" + destination.toString() + "收到了消息：\t" + tm.getText());
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    		return tm;
    	}
    
    }

## 3. 消费者监听器 ##

一般mq的消费者，都是自动消费掉生产者的业务请求的。 所以一般都配置为listener。 只有有生产者生产出业务，消费者都立即消费掉。

### 业务消费者1（QueueMessageListenerToMS.java） ###

</pre><p><pre name="code" class="java">package com.wanle.ms.mq;
    
    import javax.annotation.Resource;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    
    
    public class QueueMessageListenerToMS implements MessageListener {
    	
    	@Resource(name = "jmsTemplateToMS")
    	private JmsTemplate jmsTemplateToMS;
    	
    	@Resource(name = "jmsTemplateToLDY")
    	private JmsTemplate jmsTemplateToLDY;
    
        //当收到消息后，自动调用该方法
        @Override
        public void onMessage(Message message) {
            
            TextMessage tm = (TextMessage) message;
            try {
                System.out.println("(MS::)QueueMessageListener监听到了文本消息：\t" + tm.getText());
                //这里有个业务场景，QueueMessageListenerToMS 处理消息队列时候，会再次发送其他业务消息。
                String destination = jmsTemplateToLDY.getDefaultDestination().toString();
        		System.out.println("向队列" + destination + "发送了消息------------" + tm.getText());
        		jmsTemplateToLDY.send(new MessageCreator() {
        			public Message createMessage(Session session) throws JMSException {
        				return session.createTextMessage(tm.getText());
        			}
        		});
                //do something ...
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    
    }

### 业务消费者2 （QueueMessageListenerToLDY.java） ###

package com.wanle.ms.mq;
    
    import javax.annotation.Resource;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    
    
    public class QueueMessageListenerToLDY implements MessageListener {
    	
    	@Resource(name = "jmsTemplateToMS")
    	private JmsTemplate jmsTemplateToMS;
    	
    	@Resource(name = "jmsTemplateToLDY")
    	private JmsTemplate jmsTemplateToLDY;
    
        //当收到消息后，自动调用该方法
        @Override
        public void onMessage(Message message) {
            
            TextMessage tm = (TextMessage) message;
            try {
                System.out.println("(LDY::)QueueMessageListener监听到了文本消息：\t" + tm.getText());
                //do something ...
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    
    }

## 4.服务调用者（TestController.java） ##

package com.wanle.ms.controller.login;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    import javax.annotation.Resource;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.TextMessage;
    
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    
    import com.wanle.ms.mq.ConsumerService;
    import com.wanle.ms.mq.ProducerService;
    
    
    @Controller
    @RequestMapping("test")
    public class TestController {
    
        //队列名
        @Resource(name="toMSQueue")
        private Destination toMSQueue;
        @Resource(name="toLDYQueue")
        private Destination toLDYQueue;
    
        //队列消息生产者
        @Resource(name="producerService")
        private ProducerService producer;
        
        //队列消息消费者
        @Resource(name="consumerService")
        private ConsumerService consumer;
        
        @RequestMapping(value="/producer")
        public String producer(){
            System.out.println("------------go producer");
            Date now = new Date(); 
            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String time = dateFormat.format( now ); 
            System.out.println(time);
            producer.sendMessage(toMSQueue, time);
            return null;
        }
        
        @RequestMapping(value="/receive")
        public String queue_receive() throws JMSException {
            System.out.println("------------receive message");
            
            TextMessage tm = consumer.receive(toMSQueue);
            String message = tm.getText();
            System.out.println(message);
            return null;
        }
        
    }

# 五：测试： #

部署应用，访问localhost:8080/test/producer 即可看到效果。

以上内容为实际项目内容，绝对可靠