ActiveMQ学习总结（10）——ActiveMQ采用Spring注解方式发送和监听

对于ActiveMQ消息的发送，原声的api操作繁琐，而且如果不进行二次封装，打开关闭会话以及各种创建操作也是够够的了。那么，Spring提供了一个很方便的去收发消息的框架，spring jms。整合Spring后，代码不仅变得非常优雅，而且易用性和扩展性更好。

1. maven依赖

	org.apache.xbean
	xbean-spring
	3.16


	org.springframework
	spring-jms
	${springframework.version}


	org.apache.activemq
	activemq-all
	${activemq.version}

2.命名空间引入

3. Xml配置

第一个是配置我们的mq连接，ip+端口号，帐号密码的信息。

第二个是引入spring的mq连接池。可以配置缓存的连接数。

第三个是消息处理器，Spring默认提供了基于Jdk Serializable的消息处理和MappingJackson2MessageConventer，其实这两个挺常用，在Spring Redis中，在Spring MVC中，都有着这几种conventer的身影。

下面是两个发送消息的模版类，类似于之前讲过的RedisTemplate。向其注入上面定义的消息处理器，代码中我们会用到。（其实类中已经判断如果不进行注入就设置一个默认的，但是自己注入的话，方便我们控制）

listener-container是Spring提供的一个监听器容器，用于统一控制我们的监听类来接收处理消息。这里面有一些配置，schema有说明。可以配置响应模式，消费者数量等。开启多消费者，有助于加快队列处理速度。

4.注解方式的实现

如果要用注解的方式，就不需要在xml中自己定义消息监听容器了。只需要加入以下的代码：

这样，配置我们消费处理类上的@listener注解，即可监听对应的queue或者topic消息。

5.生产者代码

队列消息：

@Resource
@Component("queueSender")
public class QueueSender {
	@Resource(name = "jmsQueueTemplate")
	private JmsTemplate jmsQueueTemplate;// 通过@Qualifier修饰符来注入对应的bean
	public void send(String destination, final Object message) {
		jmsQueueTemplate.send(destination, new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				return jmsQueueTemplate.getMessageConverter().toMessage(message, session);
			}
		});
	}
}

订阅消息：

@Component
public class TopicSender {
	@Resource(name="jmsTopicTemplate")
	private JmsTemplate jmsTemplate;
	/**
	 * 发送一条消息到指定的队列（目标）
	 * @param queueName 队列名称
	 * @param message 消息内容
	 */
	public void publish(String destination,final Object message){
		jmsTemplate.send(destination, new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				return jmsTemplate.getMessageConverter().toMessage(message, session);
			}
		});
	}
}

6.消费者代码

package cn.test.activemq.consumer.queue;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.listener.adapter.MessageListenerAdapter;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.stereotype.Component;
import cn.test.MqBean;
import cn.test.activemq.message.types.QueueDefination;
/**
 * @author Han
 */
@Component("spqueueconsumertest")
public class SpringQueueReciverTest extends MessageListenerAdapter{
	private static final Logger log = LoggerFactory.getLogger(SpringQueueReciverTest.class);
	@JmsListener(destination=QueueDefination.TEST_QUEUE,concurrency="5-10")
	public void onMessagehehe(Message message, Session session) throws JMSException {
		try {
			MqBean bean = (MqBean) getMessageConverter().fromMessage(message);
			System.out.println(bean.getName());
			System.out.println(session);
			message.acknowledge();
			message.acknowledge();
		} catch (MessageConversionException | JMSException e) {
			e.printStackTrace();
		}
	}
}

上面的@JmsListener(destination=QueueDefination.TEST_QUEUE,concurrency="5-10")是在用注解方式监听的时候加入。如果用xml配置容易，可以忽略。

附上MqBean

public class MqBean implements Serializable{
	private Integer age;
	private String name;
	public Integer getAge() {
		return age;
	}
	public void setAge(Integer age) {
		this.age = age;
	}
	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}
}

运行效果截图：