Spring整合ActiveMQ

Spring整合ActiveMQ

相关jar包

第一个jar包肯定是ActiveMQ 的jar包了，在下载好ActiveMQ后，目录里面有一个activeMQ-all.jar的jar包

Spring相关jar包:包括基本的核心包还要有jms的包

maven 配置

<properties>
<spring.version>4.3.9.RELEASE</spring.version>
<slf4j.version>1.7.7</slf4j.version>
<log4j.version>1.2.17</log4j.version>
</properties>

<dependencies>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.4</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${spring.version}</version>
</dependency>
<!-- 添加spring-tx包 -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-tx</artifactId>
<version>${spring.version}</version>
</dependency>
<!-- 添加spring-jdbc包 -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
<version>${spring.version}</version>
</dependency>
<!-- 为了方便进行单元测试，添加spring-test包 -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${spring.version}</version>
</dependency>
<!--添加spring-web包 -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-webmvc</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-aop<
4000
/artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context-support</artifactId>
<version>${spring.version}</version>
</dependency>
<!--添加aspectjweaver包 -->
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjweaver</artifactId>
<version>1.8.5</version>
</dependency>
<!-- 添加servlet3.0核心包 -->
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>javax.servlet-api</artifactId>
<version>3.0.1</version>
</dependency>
<dependency>
<groupId>javax.servlet.jsp</groupId>
<artifactId>javax.servlet.jsp-api</artifactId>
<version>2.3.2-b01</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>2.4.2</version>
</dependency>
<!-- log start -->
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>${log4j.version}</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.35</version>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-all -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
</dependencies>

spring-ActiveMQ配置

<!-- Spring用于管理的ConnectionFactory。真实的connectionFactory由各个服务厂商提供，应该在property中引入真实的connectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory，由服务厂商提供，这里填ActiveMQ的connectionFactory 这里指向了一个连接池 -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
</bean>

用连接池：

真实connectionFactory —>connectionFactory连接池 —> Spring管理的connectionFactory

不用连接池：

真实connectionFactory —> Spring管理的connectionFactory

用连接池的情况下

<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
</bean>
<!--ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory
可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗。
当使用PooledConnectionFactory时，我们在定义一个ConnectionFactory时应该是如下定义-->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory"/>
<property name="maxConnections" value="10"/>
</bean>
<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
<property name="userName" value="admin"/>
<property name="password" value="admin"/>
</bean>

配置Spring提供的jms工具类

<!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
<!--配置默认的目的地-->
<property name="defaultDestination" ref="queueDestination"/>
</bean>

然后配置目的地

<!--这个是queue目的地，点对点的-->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<!--value的值随便填，当前目的地的name-->
<constructor-arg value="myqueue"/>
</bean>
<!--这个是pub/sub目的地，一对多的-->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="mytopic"/>
</bean>

配置监听器

<!-- 消息监听器 其实就是消息接受者，生产者-->目的地-->消费者 这个类由自己定义 必须实现javax.jms.MessageListener;的接口-->
<bean id="consumerMessageListener" class="com.mq.listener.ConsumerMessageListener"/>
<!-- 消息监听容器 消息监听容器必须配置的三个属性
1.指定connectionFactory
2.指定目的地：从哪监听消息
3.监听到消息之后交由谁去处理：指定listener 监听器-->
<bean id="jmsContainer"  class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="topicDestination" />
<property name="messageListener" ref="consumerMessageListener" />
</bean>

spring 配置就完成了

编写生产者类

这里用Spring 提供的工具类jmsTemplate发送消息

发送消息需指定目的地 和 要发送的消息

如果不指定目的地用的就是 默认的目的地

@Autowired
private JmsTemplate jmsTemplate;

public void sendMessage(Destination destination, final String message) {
System.out.println("---------------生产者发了一个消息：" + message);
jmsTemplate.send(destination, new MessageCreator(){
public Message createMessage(Session session) throws JMSException {
return  session.createTextMessage(message);
}
});
}

配置消费者

消费者类 必须实现javax.jms.MessageListener的接口

同时实现onMessage方法

public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("消息内容是：" + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}

编写测试类

记得引入junilt4 和Spring的测试包

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:applicationContext.xml")

public class TestMQ {

@Autowired
private IProducer producer;

@Autowired
@Qualifier("topicDestination")
private Destination destination;
@Test
public void testSend() throws InterruptedException {
for (int i=0; i<2; i++) {
System.out.println(destination);
producer.sendMessage(destination, "你好，生产者！这是消息：" + (i+1));

}
}
}

打开ActiveMQ 运行测试程序。

打开地址http://localhost:8161/admin/ 可查看到所有的消息