Spring整合ActiveMQ实现简单的消息队列

拼搏现实的明天。 2022-06-13 11:06 150阅读 0赞

# **Spring整合ActiveMQ实现简单的消息队列** #

## **JMS简介** ##

JMS(Java Message Service)，Java消息服务。主要用于在生产者和消费者之间进行消息传送，生产者负责生产消息，消费者负责接收处理消息。在实际的业务中，生产者生成消息，发送消息，消息会保存到ActiveMQ队列中，对应的消费者会接收消息，完成特定的业务逻辑。消息的传递模型有两种，一种是点对点，一个生产者对应一个消费者；另一种是发布订阅模式，一个生产者对应多个消费者，Java应用和ActiveMQ之间使用JMS传送消息。

## **Spring整合ActiveMQ** ##

### **引入依赖** ###

maven依赖如下

<dependency>
    	    <groupId>org.apache.activemq</groupId>
    	    <artifactId>activemq-all</artifactId>
    	    <version>5.14.5</version>
    	</dependency>	
    	<dependency>
    	    <groupId>javax.annotation</groupId>
    	    <artifactId>jsr250-api</artifactId>
    	    <version>1.0</version>
    	</dependency>
        <dependency>  
                <groupId>org.springframework</groupId>  
                <artifactId>spring-messaging</artifactId>  
                <version>${spring.version}</version>  
        </dependency>
        <dependency>  
                <groupId>org.springframework</groupId>  
                <artifactId>spring-jms</artifactId>  
                <version>${spring.version}</version>  
        </dependency>
    	<dependency>
    	    <groupId>org.apache.xbean</groupId>
    	    <artifactId>xbean-spring</artifactId>
    	    <version>4.5</version>
    	</dependency>

### **Spring中配置ActiveMQ** ###

完整配置文件为：

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
            xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
            xmlns:mvc="http://www.springframework.org/schema/mvc" xmlns:tx="http://www.springframework.org/schema/tx"
            xmlns:aop="http://www.springframework.org/schema/aop" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xmlns:jms="http://www.springframework.org/schema/jms" xmlns:amq="http://activemq.apache.org/schema/core"
            xsi:schemaLocation="http://www.springframework.org/schema/beans  
          http://www.springframework.org/schema/beans/spring-beans-4.0.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
          http://www.springframework.org/schema/tx  
          http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
          http://www.springframework.org/schema/aop  
          http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
          http://www.springframework.org/schema/jms  
          http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
          http://activemq.apache.org/schema/core
          http://activemq.apache.org/schema/core/activemq-core-5.14.5.xsd">
          
          
          <amq:connectionFactory id="activemqConnectionFactory" brokerURL="tcp://127.0.0.1:61616" userName="admin" password="admin"></amq:connectionFactory>
          
          
          <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
          	
          	<property name="targetConnectionFactory" ref="activemqConnectionFactory"></property>
          	
          	<property name="sessionCacheSize" value="100"/>
          </bean>
          
          
          <bean id="tttQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
         	
          	<constructor-arg>
          		<value>ttt.queue</value>
          	</constructor-arg>
          </bean>
          
          <bean id="tttTopicDestination" class="org.apache.activemq.command.ActiveMQTopic">
          	
          	<constructor-arg>
          		<value>ttt.topic</value>
          	</constructor-arg>
          </bean>
          
          
          
          <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
          	<property name="connectionFactory" ref="connectionFactory"/>
          	<property name="defaultDestination" ref="tttQueueDestination"/>
          	<property name="receiveTimeout" value="10000"/>
          	
          	<property name="pubSubDomain" value="false"/>
          </bean>
          
          
          <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
          	<property name="connectionFactory" ref="connectionFactory"/>
          	<property name="defaultDestination" ref="tttTopicDestination"/>
          	<property name="receiveTimeout" value="10000"/>
          	
          	<property name="pubSubDomain" value="true"/>      
          </bean>
          
          
          
          <jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
          	
          	<jms:listener destination="ttt.queue" ref="queueListener"/>
          </jms:listener-container>
          
          <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
          	
          	<jms:listener destination="ttt.topic" ref="topicListener"/>
          </jms:listener-container>
    </beans>

### **发送消息** ###

**发送Queue消息**

收到发送Queue的请求，创建队列消息发送对象，设置消息业务处理逻辑bean、数据，发送消息

@Override
    	public void sendQueueMessage() {
    		Random r = new Random();
    		Map<String, Object> map = new HashMap<String, Object>();
    		map.put("message" + r.nextInt(100), r.nextInt(10000));
    		MessageSender sender = new MessageSender("asyncProcessor.message");
    		sender.sendMessage(map);
    	}

队列消息发送对象

public class MessageSender implements Serializable {
    
    	/**
    	 * 
    	 */
    	private static final long serialVersionUID = -7611739730866180422L;
    
    	private String bean;
    
    	public MessageSender(String bean) {
    		this.bean = bean;
    	}
    
    	public String getBean() {
    		return bean;
    	}
    
    	public void sendMessage(Destination destination, Map<String, Object> map) {
    		MessageHelper.sendMessage(destination, map, getBean());
    	}
    
    	public void sendMessage(Map<String, Object> map) {
    		MessageHelper.sendMessage(map, getBean());
    	}
    
    }

调用jmsTemplate发送队列消息

@Component
    public class MessageHelper implements Serializable, InitializingBean {
    
    	/**
    	 * 
    	 */
    	private static final long serialVersionUID = 522520461829363235L;
    
    	@Autowired
    	private JmsTemplate jmsQueueTemplate;
    
    	private static JmsTemplate jmsTemplateProxy;
    
    	public JmsTemplate getJmsTemplate() {
    		return jmsQueueTemplate;
    	}
    
    	public void setJmsTemplate(JmsTemplate jmsTemplate) {
    		this.jmsQueueTemplate = jmsTemplate;
    	}
    
    	public static void sendMessage(Destination destination, final Map<String, Object> map, String processor) {
    		jmsTemplateProxy.send(destination, new MessageCreator() {
    
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				MapMessage message = session.createMapMessage();
    				message.setString("processor", processor);
    				message.setObject("data", map);
    				message.setJMSDestination(destination);
    				return message;
    			}
    		});
    	}
    
    	public static void sendMessage(Map<String, Object> map, String processor) {
    		jmsTemplateProxy.send(new MessageCreator() {
    
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				MapMessage message = session.createMapMessage();
    				message.setObject("processor", processor);
    				message.setObject("data", map);
    				return message;
    			}
    		});
    	}
    
    	@Override
    	public void afterPropertiesSet() throws Exception {
    
    		jmsTemplateProxy = this.jmsQueueTemplate;
    	}
    
    	public static JmsTemplate getJmsTemplateProxy() {
    		return jmsTemplateProxy;
    	}
    
    	public static void setJmsTemplateProxy(JmsTemplate jmsTemplateProxy) {
    		MessageHelper.jmsTemplateProxy = jmsTemplateProxy;
    	}
    }

**发送Topic消息**

收到发送Topic的请求，创建Topic消息发送对象，设置消息业务处理逻辑bean、数据，发送消息

@Override
    	public void sendTopicMessage() {
    		Random r = new Random();
    		Map<String, Object> map = new HashMap<String, Object>();
    		map.put("topic" + r.nextInt(100), r.nextInt(10000));
    		TopicSender sender = new TopicSender("asyncProcessor.topic");
    		sender.sendTopic(map);
    	}

Topic消息发送对象类

public class TopicSender implements Serializable {
    
    	/**
    	 * 
    	 */
    	private static final long serialVersionUID = -2422605328103788546L;
    
    	private String processor;
    
    	public TopicSender(String processor) {
    		this.processor = processor;
    	}
    
    	public String getProcessor() {
    		return processor;
    	}
    
    	public void setProcessor(String processor) {
    		this.processor = processor;
    	}
    
    	public void sendTopic(Destination destination, Map<String, Object> map) {
    		TopicHelper.send(destination, map, getProcessor());
    	}
    
    	public void sendTopic(Map<String, Object> map) {
    		TopicHelper.send(map, getProcessor());
    	}
    }

调用jmsTemplate发送Topic消息

@Component
    public class TopicHelper implements Serializable, InitializingBean {
    
    	/**
    	 * 
    	 */
    	private static final long serialVersionUID = 2255365346654153127L;
    
    	@Autowired
    	private JmsTemplate jmsTopicTemplate;
    
    	private static JmsTemplate jmsTopicTemplateProxy;
    
    	public static JmsTemplate getJmsTemplateProxy() {
    		return jmsTopicTemplateProxy;
    	}
    
    	public static void send(Destination destination, final Map<String, Object> map, final String processor) {
    		jmsTopicTemplateProxy.send(destination, new MessageCreator() {
    
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				MapMessage message = session.createMapMessage();
    				message.setString("processor", processor);
    				message.setObject("data", map);
    				return message;
    			}
    		});
    	}
    
    	public static void send(final Map<String, Object> map, final String processor) {
    		jmsTopicTemplateProxy.send(new MessageCreator() {
    
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				MapMessage message = session.createMapMessage();
    				message.setString("processor", processor);
    				message.setObject("data", map);
    				return message;
    			}
    		});
    	}
    
    	public JmsTemplate getJmsTopicTemplate() {
    		return jmsTopicTemplate;
    	}
    
    	public void setJmsTopicTemplate(JmsTemplate jmsTopicTemplate) {
    		this.jmsTopicTemplate = jmsTopicTemplate;
    	}
    
    	@Override
    	public void afterPropertiesSet() throws Exception {
    		jmsTopicTemplateProxy = getJmsTopicTemplate();
    	}
    }

### **监听消息** ###

**监听队列消息**

配置监听

<jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
          	
          	<jms:listener destination="ttt.queue" ref="queueListener"/>
          </jms:listener-container>

监听实现

@Component
    public class QueueListener implements MessageListener {
    
    	private static final Logger log = LoggerFactory.getLogger(MessageListener.class);
    
    	@SuppressWarnings("unchecked")
    	@Override
    	public void onMessage(Message message) {
    		MapMessage msg = (MapMessage) message;
    		try {
    			String processorName = msg.getString("processor");
    			Map<String, Object> param = (Map<String, Object>) msg.getObject("data");
    			IAsyncProcessor processor = (IAsyncProcessor) Context.getApplicationContext().getBean(processorName);
    			processor.process(param);
    		} catch (JMSException e) {
    
    			log.error("QueueListener on message error|", e);
    		}
    	}
    
    }

**监听Topic消息**

配置监听

<jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
          	
          	<jms:listener destination="ttt.topic" ref="topicListener"/>
          </jms:listener-container>

监听实现

@Component
    public class TopicListener implements MessageListener {
    
    	private static final Logger log = LoggerFactory.getLogger(MessageListener.class);
    
    	@SuppressWarnings("unchecked")
    	@Override
    	public void onMessage(Message message) {
    		MapMessage msg = (MapMessage) message;
    		try {
    			String processorName = msg.getString("processor");
    			Map<String, Object> param = (Map<String, Object>) msg.getObject("data");
    			IAsyncProcessor processor = (IAsyncProcessor) Context.getApplicationContext().getBean(processorName);
    			processor.process(param);
    		} catch (JMSException e) {
    
    			log.error("TopicListener on message error|", e);
    		}
    	}
    
    }

**业务处理**

业务逻辑处理bean实现

接口

public interface IAsyncProcessor {
    
    	/**
    	 * 逻辑处理
    	 * 
    	 * @param map
    	 */
    	public void process(Map<String, Object> map);
    }

实现

@Named("asyncProcessor.message")
    public class MessageAsyncProcessor implements IAsyncProcessor {
    
    	@Override
    	public void process(Map<String, Object> map) {
    		for (Entry<String, Object> e : map.entrySet()) {
    			System.out.println("message:" + e.getKey() + "," + e.getValue());
    		}
    
    	}
    
    }

@Named("asyncProcessor.topic")
    public class TopicAsyncProcessor implements IAsyncProcessor {
    
    	@Override
    	public void process(Map<String, Object> map) {
    		for (Entry<String, Object> e : map.entrySet()) {
    			System.out.println("topic:" + e.getKey() + "," + e.getValue());
    		}
    	}
    
    }

获取业务逻辑处理bean的context代码

@Component
    public class Context implements ApplicationContextAware {
    
    	private static ApplicationContext context;
    
    	@Override
    	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    		context = applicationContext;
    	}
    
    	public static ApplicationContext getApplicationContext() {
    		return context;
    	}
    
    }

碰到问题

引入Activemq-all新版本jar包后，activemq中的Spring部分和Spring重复，会导致spring启动报错，只需要包activemq中的spring部分删除即可以正常运行。

（完）。