ActiveMQ学习笔记

Createby：leon_zhangxf

一、概念

1. 官方网站概念：

Apache ActiveMQ is the most popular and powerful open source messaging and Integration Patterns server.

 

官网：http://activemq.apache.org/

 

2. 消息中间件概念

消息中间件是在分布式系统中完成消息的发送和接收的基础软件。

3.  ActiveMQ应用中的JMS规范名词解释

 

二、工作场景

 

三、MQ的种类

ActiveMQ、RocketMQ、RabbitMQ、ZeroMQ、HornetMQ……

 

1. MQ的对比

RabbitMQ是AMQP协议领先的一个实现，它实现了代理(Broker)架构，意味着消息在发送到客户端之前可以在中央节点上排队。此特性使得RabbitMQ易于使用和部署，适宜于很多场景如路由、负载均衡或消息持久化等，用消息队列只需几行代码即可搞定。但是，这使得它的可扩展性差，速度较慢，因为中央节点增加了延迟，消息封装后也比较大。

 

ZeroMQ是一个非常轻量级的消息系统，专门为高吞吐量/低延迟的场景开发，在金融界的应用中经常可以发现它。与RabbitMQ相比，ZeroMQ支持许多高级消息场景，但是你必须实现ZeroMQ框架中的各个块（比如Socket或Device等）。ZeroMQ非常灵活，但是你必须学习它的80页的手册（如果你要写一个分布式系统，一定要阅读它）。

 

ActiveMQ居于两者之间，类似于ZeroMQ，它可以部署于代理模式和P2P模式。类似于RabbitMQ，它易于实现高级场景，而且只需付出低消耗。它被誉为消息中间件的“瑞士军刀”。

 

内容转于：http://blog.csdn.net/chszs 作者：chszs

四、配置

1. 配置模式种类

1.1 p2p(point to point)点对点模式

 

1.2订阅、发布模式

 

2. 在Spring中继承的配置

2.1 P2P点对点模式配置（SSM框架环境）

2.1.1 消息生产者配置

spring配置文件：

<!-- 配置ActiveMQ消息中间件 -->

<!-- 消息生产者方向 -->

<!-- activemq工厂 -->

<bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

<property name="brokerURL" value="tcp://192.168.200.128:61616" />

<property name="userName" value="admin" />

<property name="password" value="admin" />

</bean>

<!-- activemq消息连接池 -->

<bean id="pooledConnectionFactoryBean" class="org.apache.activemq.pool.PooledConnectionFactoryBean">

<property name="connectionFactory" ref="activeMQConnectionFactory" />

</bean>

<!-- 将activemq工厂转换为spring工厂 -->

<bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

<property name="targetConnectionFactory" ref="pooledConnectionFactoryBean" />

</bean>

<!-- jmsTemplate -->

<bean class="org.springframework.jms.core.JmsTemplate">

<!-- 注入工厂 -->

<property name="connectionFactory" ref="singleConnectionFactory" />

<!-- 默认的mq连接目的名 -->

<property name="defaultDestinationName" value="productId" />

</bean>

 

代码部分：（这里内容是为了系统间调用Solr服务构建索引）

 

//得到更新后的数据，构建solr索引库索引

jmsTemplate.send(new MessageCreator() {

@Override

public Message createMessage(Session session) throws JMSException {

return session.createTextMessage(String.valueOf(id));

}

});

 

2.1.2 消息消费者配置

 

Spring配置内容：

<!-- 配置ActiveMQ消息中间件 -->

<!-- 消息消费者方向 -->

<!-- activemq工厂 -->

<bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

<property name="brokerURL" value="tcp://192.168.200.128:61616" />

<property name="userName" value="admin" />

<property name="password" value="admin" />

</bean>

<!-- activemq消息连接池 -->

<bean id="pooledConnectionFactoryBean" class="org.apache.activemq.pool.PooledConnectionFactoryBean">

<property name="connectionFactory" ref="activeMQConnectionFactory" />

</bean>

<!-- 将activemq工厂转换为spring工厂 -->

<bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

<property name="targetConnectionFactory" ref="pooledConnectionFactoryBean" />

</bean>

<!-- 自定义监听器 -->

<bean id="customMessageListener" class="leon.zhang.solr.service.listener.CustomMessageListener"></bean>

<!-- 配置监听器，监听消息 -->

<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">

<property name="connectionFactory" ref="singleConnectionFactory"></property>

<property name="destinationName" value="productId"/>

<!-- 自定义消息监听器 -->

<property name="messageListener" ref="customMessageListener"></property>

</bean>

 

自定义监听器：

/**

 * 自定义activemq监听器，用来监听目的消息

 * @author zxf

 *

 */

public class CustomMessageListener implements MessageListener {

 

@Autowired

private SearchService searchService;

@Override

public void onMessage(Message message) {

ActiveMQTextMessage amtm = (ActiveMQTextMessage)message;

try {

String id = amtm.getText();//获得商品id

searchService.contsructSolrIndexReferenceByProductId(Long.parseLong(id));//构建索引

} catch (Exception e) {

e.printStackTrace();

}

}

 

}

 

 

2.2 订阅、发布模式配置

2.2.1 消息生产者配置

<beans xmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:mvc="http://www.springframework.org/schema/mvc"

xmlns:context="http://www.springframework.org/schema/context"

xmlns:aop="http://www.springframework.org/schema/aop" 

xmlns:tx="http://www.springframework.org/schema/tx"

xmlns:task="http://www.springframework.org/schema/task"

xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"

xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.0.xsd

http://www.springframework.org/schema/mvc

http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd

http://www.springframework.org/schema/context

http://www.springframework.org/schema/context/spring-context-4.0.xsd

http://www.springframework.org/schema/aop

http://www.springframework.org/schema/aop/spring-aop-4.0.xsd

http://www.springframework.org/schema/tx

http://www.springframework.org/schema/tx/spring-tx-4.0.xsd

http://www.springframework.org/schema/task

   http://www.springframework.org/schema/task/spring-task-4.0.xsd

http://code.alibabatech.com/schema/dubbo        

http://code.alibabatech.com/schema/dubbo/dubbo.xsd">

<!-- 配置ActiveMQ消息中间件 -->

<!-- 消息生产者方向 -->

<!-- activemq工厂 -->

<bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

<property name="brokerURL" value="tcp://192.168.200.128:61616" />

<property name="userName" value="admin" />

<property name="password" value="admin" />

</bean>

<!-- activemq消息连接池 -->

<bean id="pooledConnectionFactoryBean" class="org.apache.activemq.pool.PooledConnectionFactoryBean">

<property name="connectionFactory" ref="activeMQConnectionFactory" />

</bean>

<!-- 将activemq工厂转换为spring工厂 -->

<bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

<property name="targetConnectionFactory" ref="pooledConnectionFactoryBean" />

</bean>

<!-- jmsTemplate -->

<bean class="org.springframework.jms.core.JmsTemplate">

<!-- 注入工厂 -->

<property name="connectionFactory" ref="singleConnectionFactory" />

<!-- 默认的mq连接目的名 -->

<property name="defaultDestinationName" value="productId" />

<!-- 设置为发布者订阅者模式 -->

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

</bean>

</beans>

 

 

2.2.2 消息消费者配置

<beans xmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:mvc="http://www.springframework.org/schema/mvc"

xmlns:context="http://www.springframework.org/schema/context"

xmlns:aop="http://www.springframework.org/schema/aop" 

xmlns:tx="http://www.springframework.org/schema/tx"

xmlns:task="http://www.springframework.org/schema/task"

xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"

xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.0.xsd

http://www.springframework.org/schema/mvc

http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd

http://www.springframework.org/schema/context

http://www.springframework.org/schema/context/spring-context-4.0.xsd

http://www.springframework.org/schema/aop

http://www.springframework.org/schema/aop/spring-aop-4.0.xsd

http://www.springframework.org/schema/tx

http://www.springframework.org/schema/tx/spring-tx-4.0.xsd

http://www.springframework.org/schema/task

   http://www.springframework.org/schema/task/spring-task-4.0.xsd

http://code.alibabatech.com/schema/dubbo        

http://code.alibabatech.com/schema/dubbo/dubbo.xsd">

<!-- 配置ActiveMQ消息中间件 -->

<!-- 消息消费者方向 -->

<!-- activemq工厂 -->

<bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

<property name="brokerURL" value="tcp://192.168.200.128:61616" />

<property name="userName" value="admin" />

<property name="password" value="admin" />

</bean>

<!-- activemq消息连接池 -->

<bean id="pooledConnectionFactoryBean" class="org.apache.activemq.pool.PooledConnectionFactoryBean">

<property name="connectionFactory" ref="activeMQConnectionFactory" />

</bean>

<!-- 将activemq工厂转换为spring工厂 -->

<bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

<property name="targetConnectionFactory" ref="pooledConnectionFactoryBean" />

</bean>

<!-- 自定义监听器 -->

<bean id="customMessageListener" class="leon.zhang.solr.service.listener.CustomMessageListener"></bean>

<!-- 配置监听器，监听消息 -->

<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">

<property name="connectionFactory" ref="singleConnectionFactory"></property>

<property name="destinationName" value="productId"/>

<!-- 自定义消息监听器 -->

<property name="messageListener" ref="customMessageListener"></property>

<!-- 设置为发布者订阅者模式 -->

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

</bean>

</beans>