成小胖学习ActiveMQ·基础篇

╰半橙微兮° 2022-04-13 12:48 161阅读 0赞

# [成小胖学习ActiveMQ·基础篇][ActiveMQ] #

过了个春节，回到公司的成小胖变成了成大胖。但是你们千万别以为他那个大肚子里面装的都是肥肉，里面的墨水也多了不少嘞，毕竟成小胖利用春节的半个月时间专心学习并研究了 ActiveMQ，嘿嘿……  
这不，为了检验下自己的学习成果，上班的第一天成小胖就去找架构师老王交流 ActiveMQ 相关的知识，还顺便向老王讨了个红包，可把成小胖给高兴坏了。  
“**来，根据你的了解说下 ActiveMQ 是什么**。”

“这个简单，ActiveMQ 是一个 MOM，具体来说是一个实现了 JMS 规范的系统间远程通信的消息代理。它……”

“**等等，先解释下什么是 MOM**。”

“好。MOM 就是面向消息中间件（Message-oriented middleware），是用于以分布式应用或系统中的异步、松耦合、可靠、可扩展和安全通信的一类软件。MOM 的总体思想是它作为消息发送器和消息接收器之间的消息中介,这种中介提供了一个全新水平的松耦合。”

“**JMS呢？**”

成小胖是个追求极致的人，为了解释得更通俗易懂，索性搬来一块白板边画边说。

“JMS 叫做 Java 消息服务（Java Message Service）,是 Java 平台上有关面向 MOM 的技术规范，旨在通过提供标准的产生、发送、接收和处理消息的 API 简化企业应用的开发，类似于 JDBC 和关系型数据库通信方式的抽象。”

![897247-20170209130542229-26011702.png][]  
“**嗯，很好。下面的这些概念你也需要特别理解下**”**：**

*  Provider：纯 Java 语言编写的 JMS 接口实现（比如 ActiveMQ 就是）
 *  Domains：消息传递方式，包括点对点（P2P）、发布/订阅（Pub/Sub）两种
 *  Connection factory：客户端使用连接工厂来创建与 JMS provider 的连接
 *  Destination：消息被寻址、发送以及接收的对象

“**你来说说这其中 P2P 和 Pub/Sub 的区别吧**”，老王给成小胖抛出了一个问题。

成小胖可不是吃素的，毕竟要是吃素的话他也吃不到这么胖……这些基本概念对他来说都是小事一桩：

P2P （点对点）消息域使用 queue 作为 Destination，消息可以被同步或异步的发送和接收，每个消息只会给一个 Consumer 传送一次。

Consumer 可以使用 MessageConsumer.receive() 同步地接收消息，也可以通过使用MessageConsumer.setMessageListener() 注册一个 MessageListener 实现异步接收。

多个 Consumer 可以注册到同一个 queue 上，但一个消息只能被一个 Consumer 所接收，然后由该 Consumer 来确认消息。并且在这种情况下，Provider 对所有注册的 Consumer 以轮询的方式发送消息。  
![897247-20170209130707869-495266263.png][]

Pub/Sub（发布/订阅，Publish/Subscribe）消息域使用 topic 作为 Destination，发布者向 topic 发送消息，订阅者注册接收来自 topic 的消息。发送到 topic 的任何消息都将自动传递给所有订阅者。接收方式（同步和异步）与 P2P 域相同。  
除非显式指定，否则 topic 不会为订阅者保留消息。当然，这可以通过持久化（Durable）订阅来实现消息的保存。这种情况下，当订阅者与 Provider 断开时，Provider 会为它存储消息。当持久化订阅者重新连接时，将会受到所有的断连期间未消费的消息。  
![897247-20170209130843776-111084090.png][]

“嗯，总结的很不错，上面的这些知识是学习 ActiveMQ 的理论基础，是必须要掌握的。”

“既然 JMS 是一个通用的规范，**那么使用它创建应用程序肯定也有一个通用的步骤吧**？”老王追问道。

“有的有的。要不您来说说这个通用步骤？就当我考考您，哈哈！”成小胖故作聪明，自以为老王作为架构师不会关注这些太具体的实现细节。

然而老王平日里亲力亲为，至今还常常撸代码，怎么会被这种小 case 所难倒？于是老王分分钟给出答案：

*  获取连接工厂
 *  使用连接工厂创建连接
 *  启动连接
 *  从连接创建会话
 *  获取 Destination
 *  创建 Producer，或
    
     *  创建 Producer
     *  创建 message
 *  创建 Consumer，或发送或接收message发送或接收 message
    
     *  创建 Consumer
     *  注册消息监听器（可选）
 *  发送或接收 message
 *  关闭资源（connection, session, producer, consumer 等)

“66666，厉害啊我的王哥！”成小胖的小聪明被老王击得粉碎！

“你嘴皮子耍够了吧，还是多动动手吧。**现在你手写上面步骤对应的代码实现吧**”，老王给了成小胖一个眼神，让他自己慢慢体会……

成小胖也不是省油的灯，马上擦干净白板，现场撸了起来（是撸代码，撸代码，撸代码，重要的事情说三遍）：

![复制代码][copycode.gif]

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();
            }
    }

![复制代码][copycode.gif]

老王满意的点点头：“还算不赖哈~ JMS 通用的规范咱们都聊完了，下面就来聊点 ActiveMQ 更具体点的东西咯。”

“好啊好啊。**要不我先基于自己的学习讲讲 ActiveMQ 的存储，您看看我哪里讲的不对或者遗漏的，可好**？”成小胖发挥了他一贯的积极主动的作风，当然内心里还是想得到老王的赞许。

“行，那就开始吧。”

ActiveMQ 在 queue 中存储 Message 时，采用先进先出顺序（FIFO）存储。同一时间一个消息被分派给单个消费者，且只有当 Message 被消费并确认时，它才能从存储中删除。

![897247-20170209150723041-822523364.png][]

对于持久化订阅者来说，每个消费者获得 Message 的副本。为了节省存储空间，Provider 仅存储消息的一个副本。持久化订阅者维护了指向下一个 Message 的指针，并将其副本分派给消费者。以这种方式实现消息存储，因为每个持久化订阅者可能以不同的速率消费 Message，或者它们可能不是全部同时运行。此外，因每个 Message 可能存在多个消费者，所以在它被成功地传递给所有持久化订阅者之前，不能从存储中删除。

![897247-20170209150745510-1362013534.png][]

“**很好，上面这段知识非常重要。其实我们可以通过表格来更清晰地展示**”，老王补充道，并在白板上画了以下表格：

<table> 
 <tbody> 
  <tr> 
   <td>消息类型</td> 
   <td>是否持久化</td> 
   <td>是否有Durable订阅者</td> 
   <td>消费者延迟启动时，消息是否保留</td> 
   <td>Broker重启时，消息是否保留</td> 
  </tr> 
  <tr> 
   <td>Queue</td> 
   <td>N</td> 
   <td>-</td> 
   <td>Y</td> 
   <td>N</td> 
  </tr> 
  <tr> 
   <td>Queue</td> 
   <td>Y</td> 
   <td>-</td> 
   <td>Y</td> 
   <td>Y</td> 
  </tr> 
  <tr> 
   <td>Topic</td> 
   <td>N</td> 
   <td>N</td> 
   <td>N</td> 
   <td>N</td> 
  </tr> 
  <tr> 
   <td>Topic</td> 
   <td>N</td> 
   <td>Y</td> 
   <td>Y</td> 
   <td>N</td> 
  </tr> 
  <tr> 
   <td>Topic</td> 
   <td>Y</td> 
   <td>N</td> 
   <td>N</td> 
   <td>N</td> 
  </tr> 
  <tr> 
   <td>Topic</td> 
   <td>Y</td> 
   <td>Y</td> 
   <td>Y</td> 
   <td>Y</td> 
  </tr> 
 </tbody> 
</table>

成小胖虽然对以上特性做过实践对比，但是并没有想到去画一个表格出来使对比更加清晰易懂。特别是当他看到老王随时就画出这个表格时便惊叹不已，大声喊道：“老王你太牛了，真是爱死你了！”

周围的同事听到后，都齐刷刷的往这边看过来。

此情此景，老王也不好意思了：“诶诶诶，说话注意哈，不要让人觉得我们在搞基。**回归正题，你再说说 ActiveMQ 常用的存储方式吧。**”

成小胖羞涩的点点头，迅速地回归原态，一五一十地说起来。

**1.KahaDB**

ActiveMQ 5.3 版本起的默认存储方式。KahaDB存储是一个基于文件的快速存储消息，设计目标是易于使用且尽可能快。它使用基于文件的消息数据库意味着没有第三方数据库的先决条件。

要启用 KahaDB 存储，需要在 activemq.xml 中进行以下配置：

![复制代码][copycode.gif]

![复制代码][copycode.gif]

**2.AMQ**

与 KahaDB 存储一样，AMQ存储使用户能够快速启动和运行，因为它不依赖于第三方数据库。AMQ 消息存储库是可靠持久性和高性能索引的事务日志组合，当消息吞吐量是应用程序的主要需求时，该存储是最佳选择。但因为它为每个索引使用两个分开的文件，并且每个 Destination 都有一个索引，所以当你打算在代理中使用数千个队列的时候，不应该使用它。

![复制代码][copycode.gif]

![复制代码][copycode.gif]

**3.JDBC**

选择关系型数据库，通常的原因是企业已经具备了管理关系型数据的专长，但是它在性能上绝对不优于上述消息存储实现。事实是，许多企业使用关系数据库作为存储，是因为他们更愿意充分利用这些数据库资源。

![复制代码][copycode.gif]

![复制代码][copycode.gif]

**4.内存存储**

内存消息存储器将所有持久消息保存在内存中。在仅存储有限数量 Message 的情况下，内存消息存储会很有用，因为 Message 通常会被快速消耗。在 activema.xml 中将 broker 元素上的 persistent 属性设置为 false 即可。

![复制代码][copycode.gif]

![复制代码][copycode.gif]

老王听完后露出赞许的笑容：“连配置都能写的这么详细，看来你确实是做了不少功课，给你点个赞。”老王终究不会吝啬自己的赞美，他也明白这些赞美对成小胖意味着什么。

成小胖得到了老王的赞赏，心里也是吃了蜜一般。

没等成小胖说话，老王拿起笔走到白板前，说：“**下面就根据我在工作中的经历，给你讲讲 ActiveMQ 的部署模式。**”

**1.单例模式**

这个就不啰嗦了，略过。

**2.无共享主从模式**

这是最简单的 Provider 高可用性的方案，主从节点分别存储 Message。从节点需要配置为连接到主节点，并且需要特殊配置其状态。

所有消息命令（消息，确认，订阅，事务等）都从主节点复制到从节点，这种复制发生在主节点对其接收的任何命令生效之前。并且，当主节点收到持久消息，会等待从节点完成消息的处理（通常是持久化到存储），然后再自己完成消息的处理（如持久化到存储）后，再返回对 Producer 的回执。

从节点不启动任何传输，也不能接受任何客户端或网络连接，除非主节点失效。当主节点失效后，从节点自动成为主节点，并且开启传输并接受连接。这是，使用 failover 传输的客户端就会连接到该新主节点。

Broker 连接配置如下：

failover://(tcp://masterhost:61616,tcp://slavehost:61616)?randomize=false

但是，这种部署模式有一些限制，

*  主节点只会在从节点连接到主节点时复制其活动状态，因此当从节点没有连接上主节点之前，任何主节点处理的 Message 或者消息确认都会在主节点失效后丢失。不过你可以通过在主节点设置 waitForSlave 来避免，这样就强制主节点在没有任何一个从节点连接上的情况下接受连接。
 *  就是主节点只能有一个从节点，并且从节点不允许再有其他从节点。
 *  把正在运行的单例配置成无共享主从，或者配置新的从节点时，你都要停止当前服务，修改配置后再重启才能生效。

![897247-20170209161210916-723854934.png][]

在可以接受一些故障停机时间的情况下，可以使用该模式。

从节点配置：

此外，可以配置 shutdownOnMasterFailure 项，表示主节点失效后安全关闭，保证没有消息丢失，允许管理员维护一个新的从节点。

**3.共享存储主从模式**

允许多个代理共享存储，但任意时刻只有一个是活动的。这种情况下，当主节点失效时，无需人工干预来维护应用的完整性。另外一个好处就是没有从节点数的限制。

有两种细分模式：

（1）基于数据库

它会获取一个表上的排它锁，以确保没有其他 ActiveMQ 代理可以同时访问数据库。其他未获得锁的代理则处于轮询状态，就会被当做是从节点，不会开启传输也不会接受连接。

![897247-20170209161329666-682001780.png][]  
（2）基于文件系统

需要获取分布式共享文件锁，linux 系统下推荐用 GFS2。

![897247-20170209161340307-1227911075.png][]