“来，根据你的了解说下 ActiveMQ 是什么。”html
“这个简单，ActiveMQ 是一个 MOM，具体来讲是一个实现了 JMS 规范的系统间远程通讯的消息代理。它……”mysql
“等等，先解释下什么是 MOM。”linux
“好。MOM 就是面向消息中间件（Message-oriented middleware），是用于以分布式应用或系统中的异步、松耦合、可靠、可扩展和安全通讯的一类软件。MOM 的整体思想是它做为消息发送器和消息接收器之间的消息中介,这种中介提供了一个全新水平的松耦合。”sql
“JMS呢？”数据库
成小胖是个追求极致的人，为了解释得更通俗易懂，索性搬来一块白板边画边说。apache
“JMS 叫作 Java 消息服务（Java Message Service）,是 Java 平台上有关面向 MOM 的技术规范，旨在经过提供标准的产生、发送、接收和处理消息的 API 简化企业应用的开发，相似于 JDBC 和关系型数据库通讯方式的抽象。”安全
“嗯，很好。下面的这些概念你也须要特别理解下”：网络
- Provider：纯 Java 语言编写的 JMS 接口实现（好比 ActiveMQ 就是）
- Domains：消息传递方式，包括点对点（P2P）、发布/订阅（Pub/Sub）两种
- Connection factory：客户端使用链接工厂来建立与 JMS provider 的链接
- Destination：消息被寻址、发送以及接收的对象
“你来讲说这其中 P2P 和 Pub/Sub 的区别吧”session
P2P （点对点）消息域使用 queue 做为 Destination，消息能够被同步或异步的发送和接收，每一个消息只会给一个 Consumer 传送一次。dom
Consumer 可使用 MessageConsumer.receive() 同步地接收消息，也能够经过使用MessageConsumer.setMessageListener() 注册一个 MessageListener 实现异步接收。
多个 Consumer 能够注册到同一个 queue 上，但一个消息只能被一个 Consumer 所接收，而后由该 Consumer 来确认消息。而且在这种状况下，Provider 对全部注册的 Consumer 以轮询的方式发送消息。
Pub/Sub（发布/订阅，Publish/Subscribe）消息域使用 topic 做为 Destination，发布者向 topic 发送消息，订阅者注册接收来自 topic 的消息。发送到 topic 的任何消息都将自动传递给全部订阅者。接收方式（同步和异步）与 P2P 域相同。
除非显式指定，不然 topic 不会为订阅者保留消息。固然，这能够经过持久化（Durable）订阅来实现消息的保存。这种状况下，当订阅者与 Provider 断开时，Provider 会为它存储消息。当持久化订阅者从新链接时，将会受到全部的断连期间未消费的消息。
“嗯，总结的很不错，上面的这些知识是学习 ActiveMQ 的理论基础，是必需要掌握的。”
“既然 JMS 是一个通用的规范，那么使用它建立应用程序确定也有一个通用的步骤吧？”
“有的有的。要不您来讲说这个通用步骤？就当我考考您，哈哈！”
答案：
- 获取链接工厂
- 使用链接工厂建立链接
- 启动链接
- 从链接建立会话
- 获取 Destination
- 建立 Producer，或
- 建立 Producer
- 建立 message
- 建立 Consumer，或发送或接收message发送或接收 message
- 建立 Consumer
- 注册消息监听器（可选）
- 发送或接收 message
- 关闭资源（connection, session, producer, consumer 等)
“如今你手写上面步骤对应的代码实现吧”
public class JMSDemo { ConnectionFactory connectionFactory; Connection connection; Session session; Destination destination; MessageProducer producer; MessageConsumer consumer; Message message; boolean useTransaction = false; try { Context ctx = new InitialContext(); connectionFactory = (ConnectionFactory) ctx.lookup("ConnectionFactoryName"); //使用ActiveMQ时：connectionFactory = new ActiveMQConnectionFactory(user, password, getOptimizeBrokerUrl(broker)); connection = connectionFactory.createConnection(); connection.start(); session = connection.createSession(useTransaction, Session.AUTO_ACKNOWLEDGE); destination = session.createQueue("TEST.QUEUE"); //生产者发送消息 producer = session.createProducer(destination); message = session.createTextMessage("this is a test"); //消费者同步接收 consumer = session.createConsumer(destination); message = (TextMessage) consumer.receive(1000); System.out.println("Received message: " + message); //消费者异步接收 consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { if (message != null) { doMessageEvent(message); } } }); } catch (JMSException e) { ... } finally { producer.close(); session.close(); connection.close(); } }
“还算不赖哈~ JMS 通用的规范我们都聊完了，下面就来聊点 ActiveMQ 更具体点的东西咯。”
“要不我先基于本身的学习讲讲 ActiveMQ 的存储，您看看我哪里讲的不对或者遗漏的，可好？”
“行，那就开始吧。”
ActiveMQ 在 queue 中存储 Message 时，采用先进先出顺序（FIFO）存储。同一时间一个消息被分派给单个消费者，且只有当 Message 被消费并确认时，它才能从存储中删除。
对于持久化订阅者来讲，每一个消费者得到 Message 的副本。为了节省存储空间，Provider 仅存储消息的一个副本。持久化订阅者维护了指向下一个 Message 的指针，并将其副本分派给消费者。以这种方式实现消息存储，由于每一个持久化订阅者可能以不一样的速率消费 Message，或者它们可能不是所有同时运行。此外，因每一个 Message 可能存在多个消费者，因此在它被成功地传递给全部持久化订阅者以前，不能从存储中删除。
“很好，上面这段知识很是重要。其实咱们能够经过表格来更清晰地展现”
| 消息类型 | 是否持久化 | 是否有Durable订阅者 | 消费者延迟启动时，消息是否保留 | Broker重启时，消息是否保留 |
| Queue | N | - | Y | N |
| Queue | Y | - | Y | Y |
| Topic | N | N | N | N |
| Topic | N | Y | Y | N |
| Topic | Y | N | N | N |
| Topic | Y | Y | Y | Y |
虽然对以上特性作过实践对比，可是并无想到去画一个表格出来使对比更加清晰易懂。
“你再说说 ActiveMQ 经常使用的存储方式吧。”
1.KahaDB
ActiveMQ 5.3 版本起的默认存储方式。KahaDB存储是一个基于文件的快速存储消息，设计目标是易于使用且尽量快。它使用基于文件的消息数据库意味着没有第三方数据库的先决条件。
要启用 KahaDB 存储，须要在 activemq.xml 中进行如下配置：
<broker brokerName="broker" persistent="true" useShutdownHook="false">
<persistenceAdapter>
<kahaDB directory="${activemq.data}/kahadb" journalMaxFileLength="16mb"/>
</persistenceAdapter>
</broker>
2.AMQ
与 KahaDB 存储同样，AMQ存储使用户可以快速启动和运行，由于它不依赖于第三方数据库。AMQ 消息存储库是可靠持久性和高性能索引的事务日志组合，当消息吞吐量是应用程序的主要需求时，该存储是最佳选择。但由于它为每一个索引使用两个分开的文件，而且每一个 Destination 都有一个索引，因此当你打算在代理中使用数千个队列的时候，不该该使用它。
<persistenceAdapter>
<amqPersistenceAdapter
directory="${activemq.data}/kahadb"
syncOnWrite="true"
indexPageSize="16kb"
indexMaxBinSize="100"
maxFileLength="10mb" />
</persistenceAdapter>
3.JDBC
选择关系型数据库，一般的缘由是企业已经具有了管理关系型数据的专长，可是它在性能上绝对不优于上述消息存储实现。事实是，许多企业使用关系数据库做为存储，是由于他们更愿意充分利用这些数据库资源。
<beans> <broker brokerName="test-broker" persistent="true" xmlns="http://activemq.apache.org/schema/core"> <persistenceAdapter> <jdbcPersistenceAdapter dataSource="#mysql-ds"/> </persistenceAdapter> </broker> <bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close"> <property name="driverClassName" value="com.mysql.jdbc.Driver"/> <property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/> <property name="username" value="activemq"/> <property name="password" value="activemq"/> <property name="maxActive" value="200"/> <property name="poolPreparedStatements" value="true"/> </bean> </beans>
4.内存存储
内存消息存储器将全部持久消息保存在内存中。在仅存储有限数量 Message 的状况下，内存消息存储会颇有用，由于 Message 一般会被快速消耗。在 activema.xml 中将 broker 元素上的 persistent 属性设置为 false 便可。
<broker brokerName="test-broker" persistent="false" xmlns="http://activemq.apache.org/schema/core"> <transportConnectors> <transportConnector uri="tcp://localhost:61635"/> </transportConnectors> </broker>
“下面就根据我在工做中的经历，给你讲讲 ActiveMQ 的部署模式。”
1.单例模式
这个就不啰嗦了，略过。
2.无共享主从模式
这是最简单的 Provider 高可用性的方案，主从节点分别存储 Message。从节点须要配置为链接到主节点，而且须要特殊配置其状态。
全部消息命令（消息，确认，订阅，事务等）都从主节点复制到从节点，这种复制发生在主节点对其接收的任何命令生效以前。而且，当主节点收到持久消息，会等待从节点完成消息的处理（一般是持久化到存储），而后再本身完成消息的处理（如持久化到存储）后，再返回对 Producer 的回执。
从节点不启动任何传输，也不能接受任何客户端或网络链接，除非主节点失效。当主节点失效后，从节点自动成为主节点，而且开启传输并接受链接。这是，使用 failover 传输的客户端就会链接到该新主节点。
Broker 链接配置以下：
failover://(tcp://masterhost:61616,tcp://slavehost:61616)?randomize=false
可是，这种部署模式有一些限制，
- 主节点只会在从节点链接到主节点时复制其活动状态，所以当从节点没有链接上主节点以前，任何主节点处理的 Message 或者消息确认都会在主节点失效后丢失。不过你能够经过在主节点设置 waitForSlave 来避免，这样就强制主节点在没有任何一个从节点链接上的状况下接受链接。
- 就是主节点只能有一个从节点，而且从节点不容许再有其余从节点。
- 把正在运行的单例配置成无共享主从，或者配置新的从节点时，你都要中止当前服务，修改配置后再重启才能生效。
在能够接受一些故障停机时间的状况下，可使用该模式。
从节点配置：
<services>
<masterConnector remoteURI="tcp://remotehost:62001" userName="Rob" password="Davies"/>
</services>
此外，能够配置 shutdownOnMasterFailure 项，表示主节点失效后安全关闭，保证没有消息丢失，容许管理员维护一个新的从节点。
3.共享存储主从模式
容许多个代理共享存储，但任意时刻只有一个是活动的。这种状况下，当主节点失效时，无需人工干预来维护应用的完整性。另一个好处就是没有从节点数的限制。
有两种细分模式：
（1）基于数据库
它会获取一个表上的排它锁，以确保没有其余 ActiveMQ 代理能够同时访问数据库。其余未得到锁的代理则处于轮询状态，就会被当作是从节点，不会开启传输也不会接受链接。
（2）基于文件系统
须要获取分布式共享文件锁，linux 系统下推荐用 GFS2。
“再讲讲我所理解的 ActiveMQ 的网络链接”
1.代理网络
支持将 ActiveMQ 消息代理连接到不一样拓扑，这就是被人们熟知的代理网络。
ActiveMQ 网络使用存储和转发的概念，其中消息老是存储在本地代理中，而后经过网络转发到另外一个代理。
当链接创建后，远程代理将把包含其全部持久和活动消费者目的地的信息传递给本地代理，本地代理根据信息决定远程代理感兴趣的 Message 并将它发送给远程代理。
若是但愿网络是双向的，您可使用网络链接器将远程代理配置为指向本地代理，或将网络链接器配置为双工，以便双向发送消息。
<networkConnectors> <networkConnector uri="static://(tcp://backoffice:61617)" name="bridge" duplex="true" conduitSubscriptions="true" decreaseNetworkConsumerPriority="false"> </networkConnector> </networkConnectors>
注意，配置的顺序很重要：
1.网络链接——须要在消息存储前创建好链接，对应 networkConnectors 元素
2.消息存储——须要在传输前配置好，对应 persistenceAdapter 元素
3.消息传输——最后配置，对应 transportConnectors 元素
2.网络发现
（1）动态发现
使用多播来支持网络动态发现。配置以下：
<networkConnectors> <networkConnector uri="multicast://default"/> </networkConnectors>
其中，multicast:// 中的默认名称表示该代理所属的组。所以使用此方式时，强烈推荐你使用一个独特的组名，避免你的代理链接到其余不相关代理。
（2）静态发现
静态发现接受代理 URI 列表，并将尝试按列表中肯定的顺序链接到远程代理。
<networkConnectors> <networkConnector uri="static:(tcp://remote-master:61617,tcp://remote-slave:61617)"/> </networkConnectors>
相关配置以下：
- initialReconnectDelay：默认值1000，表示尝试链接前的时延。
- maxReconnectDelay：默认值30000，表示链接失败后到从新创建链接之间的时延，仅在 useExponentialBackOff 启用时生效。
- useExponentialBackOff：默认值 true，若是启用，表示每次失败后增长重建链接的时延。
- backOffMultiplier：默认值2，表示启用 useExponentialBackOff 后每次的时延增量须要注意的是，网络链接将始终尝试创建到远程代理的链接。
须要注意的是，网络链接将始终尝试创建到远程代理的链接。
（3）多链接场景
当网络负载高时，使用多链接颇有意义。可是你须要确保不会重复传递消息，这能够经过过滤器来实现。
<networkConnectors> <networkConnector uri="static://(tcp://remotehost:61617)" name="queues_only" duplex="true" <excludedDestinations> <topic physicalName=">"/> </excludedDestinations> </networkConnector> <networkConnector uri="static://(tcp://remotehost:61617)" name="topics_only" duplex="true" <excludedDestinations> <queue physicalName=">"/> </excludedDestinations> </networkConnector> </networkConnectors>
上面这些知识点虽然看起来不多，但却花了不少时间看了不少英文资料，同时反复实践才理解透的。