分布式--ActiveMQ 消息中间件(二)
1. Spring+ActiveMQ
1). 导包
- Spring包：spring-aop-5.0.5.RELEASE.jar、spring-context-5.0.5.RELEASE.jar、spring-core-5.0.5.RELEASE.jar
- ActiveMQ包：activemq-all-5.15.3.jar
-
commons: commons-pool2-2.5.0.jar
图1.png
2). 项目配置
I. 配置ConnectionFactory
ConnectionFactory是用于产生到JMS服务器的链接的，Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory。
- SingleConnectionFactory对于建立JMS服务器链接的请求会一直返回同一个链接，并且会忽略Connection的close方法调用。
- CachingConnectionFactory继承了SingleConnectionFactory，所以它拥有SingleConnectionFactory的所有功能，同时它还新增了缓存功能，它可以缓存Session、MessageProducer和MessageConsumer。
ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗。
<?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:jms="http://www.springframework.org/schema/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.0.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">
<!-- 自动扫描和管理Bean配置 -->
<context:component-scan base-package="com.mazaiting"/>
<!-- 真正可以产生Connection的ConnectionFactory，由对应的JMS服务厂商提供 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
<!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗。 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory"/>
<!-- 最大连接数 -->
<property name="maxConnections" value="10"/>
</bean>
<!-- Spring 用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
</bean>
</beans>
II. 配置生产者
生产者负责产生消息并发送到JMS服务器，这通常对应的是我们的一个业务逻辑服务实现类。通常是利用Spring为我们提供的JmsTemplate类来实现的，所以配置生产者其实最核心的就是配置进行消息发送的JmsTemplate。对于消息发送者而言，在定义JmsTemplate的时候需要往里面注入一个Spring提供的ConnectionFactory对象。
<!-- Spring 提供的JMS工具类，它可以进行消息的发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个ConnectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
</bean>
利用JmsTemplate进行消息发送的时候，我们需要知道消息发送的目的地，即destination。在Jms中有一个用来表示目的地的Destination接口，它里面没有任何方法定义，只是用来做一个标识而已。当我们在使用JmsTemplate进行消息发送时没有指定destination的时候将使用默认的Destination。默认Destination可以通过在定义jmsTemplate bean对象时通过属性defaultDestination或defaultDestinationName来进行注入，defaultDestinationName对应的就是一个普通字符串。在ActiveMQ中实现了两种类型的Destination，一个是点对点的ActiveMQQueue，另一个就是支持订阅/发布模式的ActiveMQTopic。
<!-- 队列目的地，点对点 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="queue"/>
</bean>
<!-- 主题目的地，一对多 -->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic"/>
</bean>
III. 配置消费者
生产者往指定目的地Destination发送消息后，接下来就是消费者对指定目的地的消息进行消费了。通过Spring为我们封装的消息监听容器MessageListenerContainer, 负责接收信息，并把接收到的信息分发给真正的MessageListener进行处理。每个消费者对应每个目的地都需要有对应的MessageListenerContainer。通过在配置MessageConnectionFactory的时候往里面注入一个ConnectionFactory来实现的,在配置一个MessageListenerContainer的时候有三个属性必须指定，一个是表示从哪里监听的ConnectionFactory；一个是表示监听什么的Destination；一个是接收到消息以后进行消息处理的MessageListener。Spring一共为我们提供了两种类型的MessageListenerContainer，SimpleMessageListenerContainer和DefaultMessageListenerContainer。
SimpleMessageListenerContainer会在一开始的时候就创建一个会话session和消费者Consumer，并且会使用标准的JMS MessageConsumer.setMessageListener()方法注册监听器让JMS提供者调用监听器的回调函数。它不会动态的适应运行时需要和参与外部的事务管理。兼容性方面，它非常接近于独立的JMS规范，但一般不兼容Java EE的JMS限制。
大多数情况下我们还是使用的DefaultMessageListenerContainer，跟SimpleMessageListenerContainer相比，DefaultMessageListenerContainer会动态的适应运行时需要，并且能够参与外部的事务管理。它很好的平衡了对JMS提供者要求低、先进功能如事务参与和兼容Java EE环境。
<!-- 队列目的地，点对点 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="queue"/>
</bean>
<!-- 消息监听器 -->
<bean id="consumerMessageListener" class="com.mazaiting.jms.listener.ConsumerMessageListener"/>
<!-- 消息监听容器 -->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="queueDestination"/>
<property name="messageListener" ref="consumerMessageListener"/>
</bean>
IV. 完整配置文件
<?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:jms="http://www.springframework.org/schema/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.0.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">
<!-- 自动扫描和管理Bean配置 -->
<context:component-scan base-package="com.mazaiting"/>
<!-- 真正可以产生Connection的ConnectionFactory，由对应的JMS服务厂商提供 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
<!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗。 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory"/>
<!-- 最大连接数 -->
<property name="maxConnections" value="10"/>
</bean>
<!-- Spring 用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
</bean>
<!-- Spring 提供的JMS工具类，它可以进行消息的发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个ConnectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
</bean>
<!-- 队列目的地，点对点 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="queue"/>
</bean>
<!-- 消息监听器 -->
<bean id="consumerMessageListener" class="com.mazaiting.jms.listener.ConsumerMessageListener"/>
<!-- 消息监听容器 -->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="queueDestination"/>
<property name="messageListener" ref="consumerMessageListener"/>
</bean>
<!-- 主题目的地，一对多 -->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic"/>
</bean>
</beans>
3). 创建Java代码
I. 在com.mazaiting.jms.service包下创建ProducerService接口
public interface ProducerService {
public void sendMessage(Destination destination, final String message);
}
II. 在com.mazaiting.jms.service.impl包下创建ProducerServiceImpl类
@Service
public class ProducerServiceImpl implements ProducerService{
private JmsTemplate jmsTemplate;
/**
* 发送消息
* @param destination 目的地
* @param message 消息
* @throws JmsException
*/
@Override
public void sendMessage(Destination destination, final String message) {
System.out.println("-------------生产者发送消息---------------");
System.out.println("-------------生产者发了一个消息：" + message);
jmsTemplate.send(destination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}
public JmsTemplate getJmsTemplate() {
return jmsTemplate;
}
@Resource
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
}
III. 在com.mazaiting.jms.listener下创建ConsumerMessageListener类
public class ConsumerMessageListener implements MessageListener{
@Override
public void onMessage(Message message) {
// 判断是否为纯文本消息
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
System.out.println("接收到纯文本消息：");
try {
System.out.println("消息的内容是： " + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
IV. 在com.mazaiting.jms包下创建ProducerConsumerTest测试类
public class ProducerConsumerTest {
private ApplicationContext context;
private ProducerService producerService;
private Destination destination;
@Before
public void onInit() {
context = new ClassPathXmlApplicationContext("classpath:bean.xml");
producerService = (ProducerService) context.getBean("producerServiceImpl");
destination = (Destination) context.getBean("queueDestination");
}
@Test
public void sendTest() {
for (int i = 0; i < 2; i++) {
producerService.sendMessage(destination, "你好，生产者！ 这是消息" + (i+1));
}
}
}
V. 测试
- 先运行activemq.bat文件
-
在运行测试方法
图2.png
2. 消息监听器(MessageListener)
pring整合JMS的应用中我们在定义消息监听器的时候一共可以定义三种类型的消息监听器，分别是MessageListener、SessionAwareMessageListener和MessageListenerAdapter。
1). MessageListener
MessageListener是最原始的消息监听器，它是JMS规范中定义的一个接口。其中定义了一个用于处理接收到的消息的onMessage方法，该方法只接收一个Message参数。
public class ConsumerMessageListener implements MessageListener{
@Override
public void onMessage(Message message) {
// 判断是否为纯文本消息
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
System.out.println("接收到纯文本消息：");
try {
System.out.println("消息的内容是： " + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
2). SessionAwareMessageListener
SessionAwareMessageListener是Spring为我们提供的，它不是标准的JMS MessageListener。MessageListener的设计只是纯粹用来接收消息的，假如我们在使用MessageListener处理接收到的消息时我们需要发送一个消息通知对方我们已经收到这个消息了，那么这个时候我们就需要在代码里面去重新获取一个Connection或Session。SessionAwareMessageListener的设计就是为了方便我们在接收到消息后发送一个回复的消息，它同样为我们提供了一个处理接收到的消息的onMessage方法，但是这个方法可以同时接收两个参数，一个是表示当前接收到的消息Message，另一个就是可以用来发送消息的Session对象。
I. 定义ConsumerSessionAwareMessageListener
public class ConsumerSessionAwareMessageListener implements SessionAwareMessageListener<TextMessage>{
private Destination destination;
@Override
public void onMessage(TextMessage message, Session session) throws JMSException {
System.out.println("收到一条消息");
System.out.println("消息的内容是： " + message.getText());
// 创建消息生产者
MessageProducer producer = session.createProducer(destination);
// 创建文本消息
Message textMessage = session.createTextMessage("ConsumerSessionAwareMessageListener...");
// 发送消息
producer.send(textMessage);
}
public Destination getDestination() {
return destination;
}
public void setDestination(Destination destination) {
this.destination = destination;
}
}
II. 在Spring的配置文件中配置该消息监听器将处理来自一个叫sessionAwareQueue的目的地的消息，并且往该MessageListener中通过set方法注入其属性destination的值为queueDestination。这样当我们的SessionAwareMessageListener接收到消息之后就会往queueDestination发送一个消息。
bean.xml文件中添加配置
<!-- SessionAwareQueue目的地 -->
<bean id="sessionAwareQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="sessionAwareQueue"/>
</bean>
<!-- 可以获取session的MessageListener -->
<bean id="consumerSessionAwareMessageListener" class="com.mazaiting.jms.listener.ConsumerSessionAwareMessageListener">
<property name="destination" ref="queueDestination"/>
</bean>
<!-- 消息监听容器 -->
<bean id="sessionAwareListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="sessionAwareQueue"/>
<property name="messageListener" ref="consumerSessionAwareMessageListener"/>
</bean>
III. 测试类
public class ProducerConsumerTest {
private ApplicationContext context;
private ProducerService producerService;
private Destination destination;
@Before
public void onInit() {
context = new ClassPathXmlApplicationContext("classpath:bean.xml");
producerService = (ProducerService) context.getBean("producerServiceImpl");
// destination = (Destination) context.getBean("queueDestination");
destination = (Destination) context.getBean("sessionAwareQueue");
}
@Test
public void sendTest() {
for (int i = 0; i < 2; i++) {
producerService.sendMessage(destination, "你好，生产者！ 这是消息" + (i+1));
}
}
@Test
public void sessionAwareMessageListenerTest() {
producerService.sendMessage(destination, "测试SessionAwareMessageListener");
}
}
IV. 测试
图3.png
3). MessageListenerAdapter
MessageListenerAdapter类实现了MessageListener接口和SessionAwareMessageListener接口，它的主要作用是将接收到的消息进行类型转换，然后通过反射的形式把它交给一个普通的Java类进行处理。
MessageListenerAdapter会把接收到的消息做如下转换：
- TextMessage转换为String对象；
- BytesMessage转换为byte数组；
- MapMessage转换为Map对象；
- ObjectMessage转换为对应的Serializable对象。
I. 创建ConsumerListener
public class ConsumerListener {
public void handleMessage(String message) {
System.out.println("ConsumerListener通过handleMessage发送到一个纯文本消息，消息内容是：" + message);
}
public void receiveMessage(String message) {
System.out.println("ConsumerListener通过receiveMessage接收到一个纯文本消息，消息内容是：" + message);
}
}
II. bean.xml添加配置
<!-- 测试消息监听适配器的队列目的地-MessageListenerAdapter -->
<bean id="adapterQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>adapterQueue</value>
</constructor-arg>
</bean>
<!-- 适配器 -->
<bean id="messageListenerAdapter" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<property name="delegate">
<bean class="com.mazaiting.jms.listener.ConsumerListener"/>
</property>
<property name="defaultListenerMethod" value="receiveMessage"/>
</bean>
<!-- 或者使用此方式配置适配器 -->
<!-- <bean id="messageListenerAdapter" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg>
<bean class="com.mazaiting.jms.listener.ConsumerListener"/>
</constructor-arg>
</bean> -->
<!-- 消息监听适配器对应的监听容器 -->
<bean id="messageListenerAdapterContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="adapterQueue"/>
<property name="messageListener" ref="messageListenerAdapter"/>
</bean>
III. 测试代码
public class ProducerConsumerTest {
private ApplicationContext context;
private ProducerService producerService;
private Destination destination;
@Before
public void onInit() {
context = new ClassPathXmlApplicationContext("classpath:bean.xml");
producerService = (ProducerService) context.getBean("producerServiceImpl");
// destination = (Destination) context.getBean("queueDestination");
// destination = (Destination) context.getBean("sessionAwareQueue");
destination = (Destination) context.getBean("adapterQueue");
}
@Test
public void messageListenerAdapterTest() {
producerService.sendMessage(destination, "测试MessageListenerAdapter");
}
}
IV. 结果
图4.png
V. 如果我们不指定MessageListenerAdapter的defaultListenerMethod属性， 运行上述代码时控制台会输出如下结果：
图5.png
4). MessageListenerAdapter除了会自动的把一个普通Java类当做MessageListener来处理接收到的消息，另一个主要的功能是可以自动的发送返回消息。
方式一：
I. 修改生产者ProducerServiceImpl代码
@Service
public class ProducerServiceImpl implements ProducerService{
@Autowired
private JmsTemplate jmsTemplate;
@Autowired
@Qualifier("responseQueue")
private Destination responseDestination;
/**
* 发送消息
* @param destination 目的地
* @param message 消息
* @throws JmsException
*/
@Override
public void sendMessage(Destination destination, final String message) {
System.out.println("-------------生产者发送消息---------------");
System.out.println("-------------生产者发了一个消息：" + message);
jmsTemplate.send(destination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage(message);
textMessage.setJMSReplyTo(responseDestination);
return textMessage;
}
});
}
}
II. 创建ResponseQueueListener
public class ResponseQueueListener implements MessageListener{
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("发送到responseQueue的一个文本消息，内容是： " + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
III. bean.xml添加配置
<!-- 用于测试消息回复 -->
<bean id="responseQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>responseQueue</value>
</constructor-arg>
</bean>
<!-- responseQueue对应的监听器 -->
<bean id="responseQueueListener" class="com.mazaiting.jms.listener.ResponseQueueListener"/>
<!-- responseQueue对应的监听容器 -->
<bean id="responseQueueMessageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="responseQueue"/>
<property name="messageListener" ref="responseQueueListener"/>
</bean>
IV. 修改ConsumerListener
public class ConsumerListener {
public void handleMessage(String message) {
System.out.println("ConsumerListener通过handleMessage发送到一个纯文本消息，消息内容是：" + message);
}
public String receiveMessage(String message) {
System.out.println("ConsumerListener通过receiveMessage接收到一个纯文本消息，消息内容是：" + message);
return "这是ConsumerListener对象的receiveMessage方法的返回值。";
}
}
V. 执行测试方法messageListenerAdapterTest
图6.png
方式二：通过MessageListenerAdapter的defaultResponseDestination属性来指定。
I. 创建DefaultResponseQueueListener
public class DefaultResponseQueueListener implements MessageListener{
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("DefaultResponseQueueListener接收到发送到defaultResponseQueue的一个文本消息，内容是：" + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
II. bean.xml添加配置
<!-- 适配器 -->
<bean id="messageListenerAdapter" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<property name="delegate">
<bean class="com.mazaiting.jms.listener.ConsumerListener"/>
</property>
<property name="defaultListenerMethod" value="receiveMessage"/>
<!-- 添加配置 -->
<property name="defaultResponseDestination" ref="defaultResponseQueue"/>
</bean>
<!-- 默认的消息回复队列 -->
<bean id="defaultResponseQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>defaultResponseQueue</value>
</constructor-arg>
</bean>
<!-- defaultResponseQueue对应的监听器 -->
<bean id="defaultResponseQueueListener" class="com.mazaiting.jms.listener.DefaultResponseQueueListener"/>
<!-- defaultResponseQueue对应的监听容器 -->
<bean id="defaultResponseQueueMessageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="defaultResponseQueue"/>
<property name="messageListener" ref="defaultResponseQueueListener"/>
</bean>
III. 执行测试方法messageListenerAdapterTest
图7.png
5). 消息转换器MessageConverter
MessageConverter的作用主要有两方面，一方面它可以把我们的非标准化Message对象转换成我们的目标Message对象，这主要是用在发送消息的时候；另一方面它又可以把我们的Message对象转换成对应的目标对象，这主要是用在接收消息的时候。
I. 创建邮件对象
/**
* 邮件 Bean
* @author mazaiting
*/
public class Email implements Serializable{
private static final long serialVersionUID = 1L;
private String receiver;
private String title;
private String content;
public Email(String receiver, String title, String content) {
this.receiver = receiver;
this.title = title;
this.content = content;
}
public String getReceiver() {
return receiver;
}
public void setReceiver(String receiver) {
this.receiver = receiver;
}
public String getTitle() {
return title;
}
public void setTitle(String title) {
this.title = title;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append("Email [receiver=");
builder.append(receiver);
builder.append(", title=");
builder.append(title);
builder.append(", content=");
builder.append(content);
builder.append("]");
return builder.toString();
}
}
II. 创建服务接口ProducerService ，并实现
public interface ProducerService {
void sendMessage(Destination destination, Serializable obj);
}
@Service
public class ProducerServiceImpl implements ProducerService{
@Autowired
private JmsTemplate jmsTemplate;
@Override
public void sendMessage(Destination destination, Serializable obj) {
// 未使用MessageConverter
/*jmsTemplate.send(destination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createObjectMessage(obj);
}
});*/
// 使用MessageConverter, JmsTemplate就会在其内部调用预定的MessageConverter对我们的消息对象进行转换，然后再进行发送。
jmsTemplate.convertAndSend(destination, obj);
}
}
III. 创建消息转换器对象EmailMessageConverter
public class EmailMessageConverter implements MessageConverter{
@Override
public Message toMessage(Object object, Session session) throws JMSException {
return session.createObjectMessage((Serializable) object);
}
@Override
public Object fromMessage(Message message) throws JMSException, MessageConversionException {
ObjectMessage objMessage = (ObjectMessage) message;
return objMessage.getObject();
}
}```
IV. 创建消息监听器ConsumerMessageListener
```public class ConsumerMessageListener implements MessageListener{
@Override
public void onMessage(Message message) {
if (message instanceof ObjectMessage) {
ObjectMessage objMessage = (ObjectMessage) message;
try {
Object obj = objMessage.getObject();
Email email = (Email) obj;
System.out.println("接收到一个ObjectMessage,包含Email对象");
System.out.println(email.toString());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
V. 修改bean-converter.xml文件(由bean.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:jms="http://www.springframework.org/schema/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.0.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">
<!-- 自动扫描和管理Bean配置 -->
<context:component-scan base-package="com.mazaiting.jms.converter"/>
<!-- 真正可以产生Connection的ConnectionFactory，由对应的JMS服务厂商提供 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
<!-- 如果传输对象，必须设置这句为true -->
<property name="trustAllPackages" value="true"/>
</bean>
<!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗。 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory"/>
<!-- 最大连接数 -->
<property name="maxConnections" value="10"/>
</bean>
<!-- Spring 用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
</bean>
<!-- 类型转换器 -->
<bean id="emailMessageConverter" class="com.mazaiting.jms.converter.converter.EmailMessageConverter"/>
<!-- Spring 提供的JMS工具类，它可以进行消息的发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个ConnectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
<!-- 消息转换器 -->
<property name="messageConverter" ref="emailMessageConverter"/>
</bean>
<!-- 队列目的地，点对点 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="queue"/>
</bean>
<!-- 消息监听器 -->
<bean id="consumerMessageListener" class="com.mazaiting.jms.converter.listener.ConsumerMessageListener"/>
<!-- 消息监听容器 -->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="queueDestination"/>
<property name="messageListener" ref="consumerMessageListener"/>
</bean>
<!-- 测试消息监听适配器的队列目的地-MessageListenerAdapter -->
<bean id="adapterQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>adapterQueue</value>
</constructor-arg>
</bean>
<!-- 适配器 -->
<bean id="messageListenerAdapter" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<property name="delegate">
<bean class="com.mazaiting.jms.listener.ConsumerListener"/>
</property>
<property name="defaultListenerMethod" value="receiveMessage"/>
</bean>
<!-- 消息监听适配器对应的监听容器 -->
<bean id="messageListenerAdapterContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="adapterQueue"/>
<property name="messageListener" ref="messageListenerAdapter"/>
</bean>
</beans>
VI. 测试类
public class ProducerConsumerConverterTest {
private ApplicationContext context;
private ProducerService producerService;
private Destination destination;
@Before
public void onInit() {
context = new ClassPathXmlApplicationContext("classpath:bean-converter.xml");
producerService = (ProducerService) context.getBean("producerServiceImpl");
destination = (Destination) context.getBean("queueDestination");
// destination = (Destination) context.getBean("adapterQueue");
}
@Test
public void objMessageTest() {
Email email = new Email("[email protected]", "主题", "内容");
producerService.sendMessage(destination, email);
}
}
VII. 执行测试
图8.png
VIII. 修改ConsumerMessageListener代码为
public class ConsumerMessageListener implements MessageListener{
/**
* 消息转换器
*/
@Autowired
@Qualifier("emailMessageConverter")
private MessageConverter messageConverter;
@Override
public void onMessage(Message message) {
if (message instanceof ObjectMessage) {
// ObjectMessage objMessage = (ObjectMessage) message;
try {
// Object obj = objMessage.getObject();
// Email email = (Email) obj;
Email email = (Email) messageConverter.fromMessage(message);
System.out.println("接收到一个ObjectMessage,包含Email对象");
System.out.println(email.toString());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
Ⅸ. 执行结果
图9.png
Ⅹ. 使用MessageListenerAdapter做消息监听器
当我们使用MessageListenerAdapter来作为消息监听器的时候，我们可以为它指定一个对应的MessageConverter，这样Spring在处理接收到的消息的时候就会自动地利用我们指定的MessageConverter对它进行转换，然后把转换后的Java对象作为参数调用指定的消息处理方法。
- 创建ConsumerListener
public class ConsumerListener {
public void receiveMessage(String message) {
System.out.println("ConsumerListener通过receiveMessage接收到一个纯文本消息，消息内容是：" + message);
}
public void receiveMessage(Email email) {
System.out.println("接收到一个包含Email的ObjectMessage。");
System.out.println(email);
}
}
- 修改bean.converter.xml
<!-- 测试消息监听适配器的队列目的地-MessageListenerAdapter -->
<bean id="adapterQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>adapterQueue</value>
</constructor-arg>
</bean>
<!-- 适配器 -->
<bean id="messageListenerAdapter" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<property name="delegate">
<bean class="com.mazaiting.jms.converter.listener.ConsumerListener"/>
</property>
<property name="defaultListenerMethod" value="receiveMessage"/>
<!-- 配置消息转换器 -->
<property name="messageConverter" ref="emailMessageConverter"/>
</bean>
<!-- 消息监听适配器对应的监听容器 -->
<bean id="messageListenerAdapterContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="adapterQueue"/>
<property name="messageListener" ref="messageListenerAdapter"/>
</bean>
- 测试类
public class ProducerConsumerConverterTest {
private ApplicationContext context;
private ProducerService producerService;
private Destination destination;
@Before
public void onInit() {
context = new ClassPathXmlApplicationContext("classpath:bean-converter.xml");
producerService = (ProducerService) context.getBean("producerServiceImpl");
// destination = (Destination) context.getBean("queueDestination");
destination = (Destination) context.getBean("adapterQueue");
}
@Test
public void objMessageTest() {
Email email = new Email("[email protected]", "主题", "内容");
producerService.sendMessage(destination, email);
}
}
-
执行效果
图10.png
6). 事务
I. 在jmsContainer中添加sessionTransacted属性即可
<bean id="jmsContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="queueDestination" />
<property name="messageListener" ref="consumerMessageListener" />
<property name="sessionTransacted" value="true"/>
</bean>
II. 如果想接收消息和数据库访问处于同一事务中，那么我们就可以配置一个外部的事务管理同时配置一个支持外部事务管理的消息监听容器（如DefaultMessageListenerContainer）。要配置这样一个参与分布式事务管理的消息监听容器，我们可以配置一个JtaTransactionManager，当然底层的JMS ConnectionFactory需要能够支持分布式事务管理，并正确地注册我们的JtaTransactionManager。
<bean id="jmsContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="queueDestination" />
<property name="messageListener" ref="consumerMessageListener" />
<property name="transactionManager" ref="jtaTransactionManager"/>
</bean>
<bean id="jtaTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager"/>
3. Transport Connectors配置
1). Transport Connectors配置
ActiveMQ Connecor分为两种：
Transport Connector和Network Connector。transport Connector负责client和broker的连接，称为client-to-broker communication；
-
Network Connector负责broker和broker之间的连接，称为broker-to-broker communication
图11.png 从broker的角度来看，transport connector是用于接受和监听来自客户端的连接请求的一种机制。可以看到在<transportConnectors>中包含了多个<transportConnector>对象，每一个对象都定义了监听的地址、端口以及使用的协议。不同的<transportConnector>的name和uri属性值必须不同。
从客户端的角度来看，transportConnector的URI是用于和broker建立连接，以通过该连接来发送和接受消息。
2). Transport Connector的可用协议
- 传输控制协议 （TCP）
- New I/O API协议（NIO）
- 用户数据包协议（UDP）
- SSL协议（Secure Sockets Layer Protocol）
- HTTP/HTTPS协议
- 虚拟机协议（virtual machine protocol， VM）
I. 传输控制协议 （TCP）
由于TCP具有可靠传输的特性，它在ActiveMQ中也是最长使用的一种协议。在默认的配置中，TCP连接的端口为61616.TCP Connector的URI的配置语法如下：
tcp://hostname:port?key=value&key=value
tcp://hostname:port为必须部分，剩余的key/value部分为可选部分，不同的key/value用”&”符号分开,详细介绍。可选部分用于定义broker的一些额外的行为，如假设我们进行如下配置：
<transportConnector name=”tcp”uri=”tcp://localhost:61616?trace=true”/>
则broker会在日志中记录经过这个connector所发送的每一条消息，这对于调试会有很大的帮助。
II. New I/O API协议（NIO）
NIO是在Java SE 1.4开始引进的，用于提供一种网络编程和访问现代操作系统的底层I/O.NIO中使用最多的两种特性是selector和non-blocking I/O，它们使得开发人员可以使用相同的资源来处理更多的网络连接（如果不使用NIO技术，系统将无法承受这么多的网络连接）。从客户端的角度来看，NIO transport connector与TCP connector是相同的，这是因为NIO connector实际上采用TCP来传输数据。唯一不同的是，NIO transport connector是利用NIO API实现的。这使得NIO在以下几种情况下更为适合：
- 大量的客户端需要连接broker。通常，与broker之间的连接数是受操作系统的线程数限制的，由于相比TCP connector而言，NIO connector会占用更少的线程。所以，当TCP connector无法满足要求时，可以考虑使用NIO进行替代。
- broker之间存在严重的网路阻塞。NIO connector会比TCP connector拥有更好的性能，因为它会使用更少的资源。
NIO的配置语法与TCP的配置相似，只需将tcp替换成nio即可：
nio://hostname:port?key=value
NIO的默认监听端口是61618，它也有trace这个可以，用于决定是否记录所有传输的消息。
III. 用户数据包协议（UDP）
根据UDP的特性，它一般用于要求传出速率快，且能够忍受丢包发生的情况。你可以使用UDP协议来连接ActiveMQ的UDP connector。它的URI语法与TCP及NIO基本一致：
udp://hostname:port?key=value
详细内容,通常以下两种情况可以考虑使用UDP：
- broker有防火墙的保护，只能通过UDP端口来访问；
- 消息传送对实时性要求非常高，可以通过UDP来尽可能的消除网络延迟；
由于UDP的不可靠特性，所以在使用时，必须考虑丢包时的处理。
IV. SSL协议（Secure Sockets Layer Protocol）
当broker暴露于不安全的网络环境，而你的数据需要保密时，可以使用SSL。它是用于通过TCP传输加密数据的一种协议，它通过一对钥匙（公钥和私钥）来保证一条安全通道。ActiveMQ提供SSL connector，它的语法如下：
ssl://hostname:port?key=value
由于ssl采用TCP进行传输，它的选项跟TCP是一致的，当然如果你需要更加详细的说明，可以参考：http://activemq.apache.org/ssl-transport-reference.html，另外需要说明的是ssl connector默认监听61617端口。与前3种不同，ssl connector还需要ssl证书来保证其正确运行。JSSE提供了两类文件分别用于存储key和certification，用于存储key的文件称为keystores，存储certification的称为truststores.
V. HTTP/HTTPS协议
有些服务器的防火墙设置了只能提供HTTP访问，这时便是HTTP/HTTPS connector的地盘了。ActiveMQ实现的HTTP/HTTPS协议用于传输XML数据，它们的配置语法分别如下：
http://hostname:port?key=value
https://hostname:port?key=value
还需要添加activem1-optional包.
VI. 虚拟机协议（virtual machine protocol， VM）
ActiveMQ还可以嵌入Java应用程序中，这样Java程序与broker之间的连接不会经过网络，而是在同一个JVM中进行，从而可以减少网络传输，大幅提升性能。这种方式成为虚拟机协议。使用VM协议的broker拥有与其它协议相同的功能，VM connector的配置语法如下：
vm://brokerName?key=value
3). network connectors部署集群
I. 网络连接模式
针对海量消息所要求的横向扩展性和系统的高可用性，ActiveMQ提供了网络连接模式的集群功能。简单的说，就是通过把多个不同的broker实例连接在一起，作为一个整体对外提供服务，从而提高整体对外的消息服务能力。通过这种方式连接在一起的broker实例之间，可以共享队列和消费者列表，从而达到分布式队列的目的。
II. 拓扑结构
几种不同的ActiveMQ部署拓扑结构（嵌入、主从复制、网络连接）：
图12.png
III. 配置示例
在activemq.xml的broker节点内添加：
<networkConnectors>
<networkConnectoruri=“static:(tcp://localhost:62001)”/>
</networkConnectors>
uri也可以使用其他两种方式：
multicast://default
masterslave:(tcp://host1:61616,tcp://host2:61616,tcp://..)
IV. 静态URI配置
使用静态URI方式可以指定多个URL，networkConnector将连接到每一个broker。
<networkConnectors>
<networkConnector uri="static:(tcp://host1:61616,tcp://host2:61616,tcp://..)"/>
</networkConnectors>
URI的几个属性：
| 属性 | 默认值 | 描述 |
|---|---|---|
| initialReconnectDelay | 1000 | 重连之前等待的时间(ms) (如果useExponentialBackOff为 false) |
| maxReconnectDelay | 30000 | 重连之前等待的最大时间(ms) |
| useExponentialBackOff | true | 每次重连失败时是否增大等待时间 |
| backOffMultiplier | 2 | 增大等待时间的系数 |