ActiveMQ集成Spring使用

ActiveMQ的安装（忽略），主要说明其集成Spring配置使用（点对点模式）。

一、创建Mavne项目，所需ActiveMQ依赖包：

<dependency>

   <groupId>org.apache.activemq</groupId>

   <artifactId>activemq-client</artifactId>

   <version>5.14.5</version>

</dependency>

<dependency>

   <groupId>org.apache.activemq</groupId>

   <artifactId>activemq-pool</artifactId>

   <version>5.14.5</version>

</dependency>

<dependency>

   <groupId>org.springframework</groupId>

   <artifactId>spring-jms</artifactId>

   <version>4.1.9.RELEASE</version>

</dependency>

忽略Spring的集成

二、配置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:aop="http://www.springframework.org/schema/aop"
       xmlns:tx="http://www.springframework.org/schema/tx"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:p="http://www.springframework.org/schema/p"  
       xmlns:cache="http://www.springframework.org/schema/cache"
       xsi:schemaLocation="
            http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
            http://www.springframework.org/schema/tx 
            http://www.springframework.org/schema/tx/spring-tx-4.1.xsd
            http://www.springframework.org/schema/aop 
            http://www.springframework.org/schema/aop/spring-aop-4.1.xsd
            http://www.springframework.org/schema/context      
            http://www.springframework.org/schema/context/spring-context-4.1.xsd 
            http://www.springframework.org/schema/cache 
            http://www.springframework.org/schema/cache/spring-cache-4.1.xsd">
    
    <context:annotation-config />
    <context:component-scan base-package="com.test.mq" />
        
    <bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    	<property name="brokerURL" value="tcp://127.0.0.1:61616" />
    </bean>
    
    <!-- 配置JMS连接工长 -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <constructor-arg ref="amqConnectionFactory" />
        <property name="sessionCacheSize" value="100" />
    </bean>
    
    <!-- 定义消息队列（Queue） -->
    <bean id="demoQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <!-- 设置消息队列的名字 -->
        <constructor-arg>
            <value>mq.demo</value>
        </constructor-arg>
    </bean>
    
    <!-- 配置JMS模板（Queue），Spring提供的JMS工具类，它发送、接收消息。 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="demoQueueDestination" />
        <property name="receiveTimeout" value="10000" />
        <!-- true是topic，false是queue，默认是false，此处显示写出false -->
        <property name="pubSubDomain" value="false" />
    </bean>
    
    <bean id="queueMessageListener" class="com.test.mq.ActiveMqListener" />
    <!-- 显示注入消息监听容器（Queue），配置连接工厂，监听的目标是demoQueueDestination，监听器是上面定义的监听器 -->
    <bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="demoQueueDestination" />
        <property name="messageListener" ref="queueMessageListener" />
    </bean>
    
 </beans>

 三、Java代码

package com.test.mq;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.apache.log4j.Logger;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;

@Service
public class ActiveMqSender {
	private static Logger logger = Logger.getLogger(ActiveMqSender.class.getName());

	@Resource
	private JmsTemplate jmsTemplate;

	/**
	 * 向指定队列发送消息
	 */
	public void sendMessage(Destination destination, final String msg) {
		jmsTemplate.send(destination, new MessageCreator() {
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage(msg);
			}
		});
		logger.info("向队列" + destination.toString() + "发送了消息------------" + msg);
	}

	/**
	 * 向默认队列发送消息
	 */
	public void sendMessage(final String msg) {
		String destination = jmsTemplate.getDefaultDestination().toString();
		jmsTemplate.send(new MessageCreator() {
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage(msg);
			}
		});
		logger.info("向队列" + destination + "发送了消息------------" + msg);
	}

}

 

package com.test.mq;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;

public class ActiveMqListener implements MessageListener {
	private static Logger logger = Logger.getLogger(ActiveMqListener.class.getName());

	@Override
	public void onMessage(Message message) {
		TextMessage tm = (TextMessage) message;
		try {
			logger.info("ActiveMqListener监听到了文本消息：\t" + tm.getText());
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

}

 四、测试

@Controller
@RequestMapping("/common")
public class CommonController {
	@Resource
	private ActiveMqSender activeMqSender;
	
	@RequestMapping(value = "testMq")
	@ResponseBody
	public String testMq(int len) {
		for(int i=0;i<len;i++){
			activeMqSender.sendMessage("message content "+i);
		}
		return "ok";
	}
}

 启动项目，启动MQ服务，访问测试方法，在MQ中可查看队列处理情况：

 

补充：

Java消息服务（Java Message Service，JMS）规范目前支持两种消息模型：点对点（point to point， queue）和发布/订阅（publish/subscribe，topic）。

区别：点对点模式不可重复消费；发布/订阅模式可以重复消费。

传统企业型消息队列ActiveMQ遵循了JMS规范，实现了点对点和发布订阅模型，但其他流行的消息队列RabbitMQ（消费模式为推push）、Kafka（消费模式为拉pull）并没有遵循JMS规范。

 

ActiveMQ Prefetch Limit

Default Prefetch Limit（默认预取限制）：不同的消费者类型有不同的默认设置，具体设置如下：

Queue consumer：默认1000

如果你使用一组消费者进行分散工作量的话（一个Queue对应多个消费者），典型的你应该把数字设置的小一些。如果一个消费者被允许可以聚集大量的未被确认的消息的话，会导致其它的消费者无事可做。同时，如果这个消费者出错的话，会导致大量的消息不能被处理，直到消费者恢复之前。

Queue browser：默认500

Topic consumer：默认32766

默认值32766是数字short的最大值，也是预取限制的最大值。

Durable topic subscriber：默认100

 

Per broker：你可以设置连接Broker的所有消费者的预取限制，通过设置borker的目标策略。设置目标策略需要在broker中增加子条目 destinationPolicy 。参考如下：

<broker ... >

  ... 

  <destinationPolicy> 

    <policyMap> 

      <policyEntries> 

        <policyEntry queue="queue.>" queuePrefetch=”1”/> 

        <policyEntry topic="topic.>" topicPrefetch=”1000”/> 

      </policyEntries> 

    </policyMap> 

  </destinationPolicy> 

  ... 

</broker>

在前面的例子中，所有开头以 queue命名的queue的预取限制设置为1.（>是一个通配符，用于匹配一个或者多个命名段）。所有开头以 topic命名topic的预取限制设置为1000.

消费模式：prefetch预取，即push，当prefetch设置为0，表示消费者主动pull消息。