Spring与ActiveMQ整合

短命女 2022-09-25 14:19 130阅读 0赞

maven依赖：

<dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jms</artifactId>
        <version>4.3.2.RELEASE</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.xbean</groupId>
        <artifactId>xbean-spring</artifactId>
        <version>4.5</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-core</artifactId>
        <version>5.7.0</version>
    </dependency>

activemq.properties内容：

brokerURL=tcp://127.0.0.1:61616
    activemq.userName=admin
    activemq.password=admin
    queue.address=queue
    topic.address=topic

spring与activemq整合的文件，由于此文件我是直接在spring与spring MVC整合的文件spring-mvc.xml中引入的，我已经在spring-mvc.xml中全局扫描含有注解的文件了，所以在这不需要扫描有注解的文件了：

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
    	xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:amq="http://activemq.apache.org/schema/core"
    	xmlns:jms="http://www.springframework.org/schema/jms"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans   
            http://www.springframework.org/schema/beans/spring-beans-4.2.xsd   
            http://www.springframework.org/schema/context   
            http://www.springframework.org/schema/context/spring-context-4.2.xsd
            http://www.springframework.org/schema/jms
            http://www.springframework.org/schema/jms/spring-jms-4.2.xsd
            http://activemq.apache.org/schema/core
            http://activemq.apache.org/schema/core/activemq-core-5.7.0.xsd">
    
        
        <bean id="activemqPropertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">  
            <property name="order" value="3" />
            <property name="ignoreUnresolvablePlaceholders" value="true" />
            <property name="locations">
    			<list>
    				<value>classpath:activemq.properties</value>
    			</list>
    		</property>
        </bean>
        
        
        
        
    
    	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    	   <property name="brokerURL" value="${brokerURL}"/>
    	   <property name="userName" value="${activemq.userName}"/>
    	   <property name="password" value="${activemq.password}"/>
    	</bean>
    	
        
          
        
        <bean id="jmsconnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
              
            <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
            
            <property name="sessionCacheSize" value="20" />
        </bean>
        
        
        
    	<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
    	    <constructor-arg value="${queue.address}"/>
    	</bean>
    	
    	<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
    	    <constructor-arg value="${topic.address}"/>
    	</bean>
        
        
        
        <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
               
            <property name="connectionFactory" ref="jmsconnectionFactory"/>
            <property name="defaultDestination" ref="queueDestination" />
            
            
            
            <property name="pubSubDomain" value="false" />
        </bean>
    
        
        <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
               
            <property name="connectionFactory" ref="jmsconnectionFactory"/>
            <property name="defaultDestination" ref="topicDestination" />
            
            
            
            <property name="pubSubDomain" value="true" />
        </bean>
        
        
        
        
        <jms:listener-container destination-type="queue" container-type="default" connection-factory="jmsconnectionFactory" acknowledge="auto">
            
            <jms:listener destination="${queue.address}" ref="queueConsumer1"/>
            <jms:listener destination="${queue.address}" ref="queueConsumer2"/>
        </jms:listener-container>
    
        
        <jms:listener-container destination-type="topic" container-type="default" connection-factory="jmsconnectionFactory" acknowledge="auto">
            
            <jms:listener destination="${topic.address}" ref="topicConsumer1"/>
            <jms:listener destination="${topic.address}" ref="topicConsumer2"/>
        </jms:listener-container>
        
    </beans>

消息生产者文件的内容：

QueueProducer.java文件：

package com.cn.common.jmsProducer;
    
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Component;
    @Component
    public class QueueProducer {
    
    	@Autowired
    	@Qualifier("jmsQueueTemplate")
    	private JmsTemplate jmsTemplate; //注入配置文件中的bean
    	
    	@Autowired
    	@Qualifier("queueDestination")  
        private Destination destination;
    	
    	public void send(final String message){
    		/*jmsTemplate.send(destination, new MessageCreator() {
    			@Override
    			public Message createMessage(Session paramSession) throws JMSException {
    				return paramSession.createTextMessage(message);
    			}
    		});*/
    		jmsTemplate.send(new MessageCreator() {
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				return session.createTextMessage(message);
    			}
    		});;
    	}
    }

TopicProducer.java文件：

package com.cn.common.jmsProducer;
    
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Component;
    @Component
    public class TopicProducer {
    
    	@Autowired
    	@Qualifier("jmsTopicTemplate")
    	private JmsTemplate jmsTemplate; //注入配置文件中的bean
    	
    	@Autowired
    	@Qualifier("topicDestination")
        private Destination destination;
    	
    	public void send(final String message){
    		/*jmsTemplate.send(destination, new MessageCreator() {
    			@Override
    			public Message createMessage(Session paramSession) throws JMSException {
    				return paramSession.createTextMessage(message);
    			}
    		});*/
    		jmsTemplate.send(new MessageCreator() {
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				return session.createTextMessage(message);
    			}
    		});;
    	}
    }

消息消费者文件的内容：

QueueConsumer1.java文件：

package com.cn.common.jmsConsumer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.springframework.stereotype.Component;
    
    @Component
    public class QueueConsumer1 implements MessageListener{
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			TextMessage text = null;
    			if(message instanceof TextMessage){
    				text = (TextMessage) message;
    				System.out.println("QueueConsumer1接收到消息："+text.getText());
    			}
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    
    }

QueueConsumer2.java文件：

package com.cn.common.jmsConsumer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.springframework.stereotype.Component;
    
    @Component
    public class QueueConsumer2 implements MessageListener{
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			TextMessage text = null;
    			if(message instanceof TextMessage){
    				text = (TextMessage) message;
    				System.out.println("QueueConsumer2接收到消息："+text.getText());
    			}
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    
    }

TopicConsumer1.java文件：

package com.cn.common.jmsConsumer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.springframework.stereotype.Component;
    @Component
    public class TopicConsumer1 implements MessageListener{
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			TextMessage text = null;
    			if(message instanceof TextMessage){
    				text = (TextMessage) message;
    				System.out.println("TopicConsumer1接收到消息："+text.getText());
    			}
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    
    }

TopicConsumer2.java文件：

package com.cn.common.jmsConsumer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.springframework.stereotype.Component;
    @Component
    public class TopicConsumer2 implements MessageListener{
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			TextMessage text = null;
    			if(message instanceof TextMessage){
    				text = (TextMessage) message;
    				System.out.println("TopicConsumer2接收到消息："+text.getText());
    			}
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    
    }

下面就是在我们需要发送消息的java文件中依赖注入QueueProducer发送点对点消息或TopicProducer发送发布订阅主题消息了。