ActiveMQ+Spring工程创建详解（附工程文件）

ActiveMQ是Apache所提供的一个开源的消息系统，完全采用Java来实现，因此，它能很好地支持J2EE提出的JMS（Java Message Service,即Java消息服务）规范。JMS是一组Java应用程序接口，它提供消息的创建、发送、读取等一系列服务。JMS提供了一组公共应用程序接口和响应的语法，类似于Java数据库的统一访问接口JDBC,它是一种与厂商无关的API，使得Java程序能够与不同厂商的消息组件很好地进行通信。

JMS支持两种消息发送和接收模型。一种称为P2P(Ponit to Point)模型，即采用点对点的方式发送消息。P2P模型是基于队列的，消息生产者发送消息到队列，消息消费者从队列中接收消息，队列的存在使得消息的异步传输称为可能，P2P模型在点对点的情况下进行消息传递时采用。

另一种称为Pub/Sub(Publish/Subscribe，即发布-订阅)模型，发布-订阅模型定义了如何向一个内容节点发布和订阅消息，这个内容节点称为topic(主题)。主题可以认为是消息传递的中介，消息发布这将消息发布到某个主题，而消息订阅者则从主题订阅消息。主题使得消息的订阅者与消息的发布者互相保持独立，不需要进行接触即可保证消息的传递，发布-订阅模型在消息的一对多广播时采用。

ActiveMQ的安装

下载最新的安装包apache-activemq-5.13.2-bin.tar.gz

下载之后解压： tar -zvxf apache-activemq-5.13.2-bin.tar.gz

ActiveMQ目录内容有：

• bin目录包含ActiveMQ的启动脚本
• conf目录包含ActiveMQ的所有配置文件
• data目录包含日志文件和持久性消息数据
• example: ActiveMQ的示例
• lib: ActiveMQ运行所需要的lib
• webapps: ActiveMQ的web控制台和一些相关的demo

ActiveMQ的默认服务端口为61616，这个可以在conf/activemq.xml配置文件中修改：

<transportConnectors>
    <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
    <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
</transportConnectors>

JMS的规范流程

1. 获得JMS connection factory. 通过我们提供特定环境的连接信息来构造factory。
2. 利用factory构造JMS connection
3. 启动connection
4. 通过connection创建JMS session.
5. 指定JMS destination.
6. 创建JMS producer或者创建JMS message并提供destination.
7. 创建JMS consumer或注册JMS message listener.
8. 发送和接收JMS message.
9. 关闭所有JMS资源，包括connection, session, producer, consumer等。

案例（整合Spring）

pom.xml

<dependency>
  <groupId>junit</groupId>
  <artifactId>junit</artifactId>
  <version>4.11</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-all</artifactId>
  <version>5.11.1</version>
</dependency>
<dependency>
  <groupId>org.springframework</groupId>
  <artifactId>spring-jms</artifactId>
  <version>4.1.4.RELEASE</version>
</dependency>
<dependency>
  <groupId>org.springframework</groupId>
  <artifactId>spring-test</artifactId>
  <version>4.1.4.RELEASE</version>
</dependency>

Queue类型消息

spring配置文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd">
  <!-- 配置JMS连接工厂 -->
  <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <!-- ActiveMQ服务的地址和端口-->
    <property name="brokerURL" value="failover:(tcp://192.168.147.131:61616)" />
  </bean>
  <!-- 定义消息队列（Queue） -->
  <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
    <!-- 设置消息队列的名字 -->
    <constructor-arg>
      <value>testSpringQueue</value>
    </constructor-arg>
  </bean>
<!-- 定义消息发布（Pub/Sub） -->
<!-- 	<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"> -->
<!-- 		<constructor-arg> -->
<!-- 			<value>topic</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="queueDestination" />
    <property name="receiveTimeout" value="10000" />
  </bean>
</beans>

推送代码

package com.mq.spring.queue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import javax.annotation.Resource;
import javax.jms.*;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations={"classpath:spring-mq-queue.xml"})
public class QueueSender {
  @Resource
  private JmsTemplate jmsTemplate;
  @Test
  public void send(){
    sendMqMessage(null,"spring activemq queue type message !");
  }
  /**
   * 说明:发送的时候如果这里没有显示的指定destination.将用spring xml中配置的destination
   * @param destination
   * @param message
   */
  public void sendMqMessage(Destination destination, final String message){
    if(null == destination){
      destination = jmsTemplate.getDefaultDestination();
    }
    jmsTemplate.send(destination, new MessageCreator() {
      public Message createMessage(Session session) throws JMSException {
        return session.createTextMessage(message);
      }
    });
    System.out.println("spring send message...");
  }
  public void setJmsTemplate(JmsTemplate jmsTemplate) {
    this.jmsTemplate = jmsTemplate;
  }
}

消费代码

package com.mq.spring.queue;
import org.junit.Test;
import javax.jms.*;
import org.junit.runner.RunWith;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import javax.annotation.Resource;
import javax.jms.Message;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations={"classpath:spring-mq-queue.xml"})
public class QueueReceiver {
  @Resource
  private JmsTemplate jmsTemplate;
  @Test
  public void receiveMqMessage(){
    Destination destination = jmsTemplate.getDefaultDestination();
    receive(destination);
  }
  /**
   * 接受消息
   */
  public void receive(Destination destination) {
    TextMessage tm = (TextMessage) jmsTemplate.receive(destination);
    try {
      System.out.println("从队列" + destination.toString() + "收到了消息：\t" + tm.getText());
    } catch (JMSException e) {
      e.printStackTrace();
    }
  }
  public void setJmsTemplate(JmsTemplate jmsTemplate) {
    this.jmsTemplate = jmsTemplate;
  }
}

说明:上面的生产者和消费者使用同一套配置文件,使用独立的程序去接收消息,spring jms也提供了消息监听处理.接下来我们换成监听式消费

配置文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd">
  <!-- 配置JMS连接工厂 -->
  <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="failover:(tcp://192.168.147.131:61616)" />
  </bean>
  <!-- 定义消息队列（Queue） -->
  <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
    <!-- 设置消息队列的名字 -->
    <constructor-arg>
      <value>testSpringQueue</value>
    </constructor-arg>
  </bean>
  <!-- 配置消息队列监听者（Queue） -->
  <bean id="consumerMessageListener" class="com.mq.spring.queue.ConsumerMessageListener" />
  <!-- 消息监听容器（Queue），配置连接工厂，监听的队列是testSpringQueue，监听器是上面定义的监听器 -->
  <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>
</beans>

监听器代码

public class ConsumerMessageListener implements MessageListener{
  @Override
  public void onMessage(Message message) {
    TextMessage tm = (TextMessage) message;
    try {
      System.out.println("ConsumerMessageListener收到了文本消息：\t" + tm.getText());
    } catch (JMSException e) {
      e.printStackTrace();
    }
  }
}

这样我们的消息消费就可以在监听器中处理消费了.生产的代码不变,修改发送者的消息体内容,执行生产程序

Topic类型消息

在使用 Spring JMS的时候，主题（ Topic）和队列消息的主要差异体现在JmsTemplate中 “pubSubDomain”是否设置为 True。如果为 True，则是 Topic；如果是false或者默认，则是 queue。

<property name="pubSubDomain" value="true" />

topic类型消费配置文件说明

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd">
  <!-- 配置JMS连接工厂 -->
  <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="failover:(tcp://192.168.147.131:61616)" />
  </bean>
  <!-- 定义消息Destination -->
  <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
    <!-- 设置消息队列的名字 -->
    <constructor-arg>
      <value>testSpringTopic</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="destination" />
    <property name="receiveTimeout" value="10000" />
  </bean>
  <!-- 配置消息消费监听者 -->
  <bean id="consumerMessageListener" class="com.mq.spring.topic.ConsumerMessageListener" />
  <!-- 消息监听容器，配置连接工厂,监听器是上面定义的监听器 -->
  <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="destination" ref="destination" />
    <!--主题（Topic）和队列消息的主要差异体现在JmsTemplate中"pubSubDomain"是否设置为True。如果为True，则是Topic；如果是false或者默认，则是queue-->
    <property name="pubSubDomain" value="true" />
    <property name="messageListener" ref="consumerMessageListener" />
  </bean>
</beans>

生产者代码

package com.mq.spring.topic;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations={"classpath:spring-mq-topic.xml"})
public class TopicSender {
  @Resource
  private JmsTemplate jmsTemplate;
  @Test
  public void send(){
    sendMqMessage(null,"spring activemq topic type message[with listener] !");
  }
  /**
   * 说明:发送的时候如果这里没有显示的指定destination.将用spring xml中配置的destination
   * @param destination
   * @param message
   */
  public void sendMqMessage(Destination destination, final String message){
    if(null == destination){
      destination = jmsTemplate.getDefaultDestination();
    }
    jmsTemplate.send(destination, new MessageCreator() {
      public Message createMessage(Session session) throws JMSException {
        return session.createTextMessage(message);
      }
    });
    System.out.println("spring send message...");
  }
  public void setJmsTemplate(JmsTemplate jmsTemplate) {
    this.jmsTemplate = jmsTemplate;
  }
}

监听器代码

public class ConsumerMessageListener implements MessageListener{
  @Override
  public void onMessage(Message message) {
    TextMessage tm = (TextMessage) message;
    try {
      System.out.println("ConsumerMessageListener收到了文本消息：\t" + tm.getText());
    } catch (JMSException e) {
      e.printStackTrace();
    }
  }
}

参考资料

• http://www.importnew.com/19767.html

• http://blog.csdn.net/u013256816/article/details/51161548

• http://www.tuicool.com/articles/FVf26b

• http://haohaoxuexi.iteye.com/blog/1893038

• Spring+ActiveMQ+Mysql 配置JMS

  ​