初识消息中间件之 == ActiveMQ

一、消息队列概述

　　消息（Message）是指在应用间传送的数据。消息可以非常简单，比如只包含文本字符串，也可以更复杂，可能包含嵌入对象。

　　消息队列（Message Queue）是一种应用间的通信方式，消息发送后可以立即返回，由消息系统来确保消息的可靠传递。消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的。这样发布者和使用者都不用知道对方的存在。

　　消息队列中间件是分布式系统中重要的组件，主要解决异步消息，流量削峰，应用耦合等问题。实现高性能，高可用，可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。目前使用较多的消息队列产品有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ等。

生活中的例子

老式餐厅点餐后需呆在座位上等餐，中途不能离开去干别的事，如果离开去干别的事，餐好了，点餐人却不知道。

新式餐厅点餐后，餐厅会提供一个“电子盘”给顾客，顾客可以不用在店里等餐，可以去附近逛逛，买买东西，等餐好了，手上的“电子盘”就会响，通知顾客可以回去就餐了。

对比以上两种形式，第二种情形就像消息队列一样，点完餐以后就可以去处理别的事情，不用一直在餐厅等着。

二、消息队列的作用

上面说了消息队列主要解决了异步处理，流量削峰，应用耦合等三个方面的问题。

异步处理

场景说明：用户注册后，系统要发送注册邮件和注册短信。传统的方式有两种，串行模式和并行方式。

串行模式：将注册信息存入数据库成功后，先发送注册邮件再发送注册短信，以上三个步骤都完成后，将成功的信息返回给客户端。

初识消息中间件之 ==> ActiveMQ

并行模式：将注册信息存入数据库成功后，发送邮件的同时发送注册短信，以上三个任务都完成后返回给客户端，与串行模式的差别是并行模式可以提高处理的时间。

初识消息中间件之 ==> ActiveMQ

假设每个业务结点的处理时间为50ms,不考虑网络开销，则串行模式的时间为150ms，并行模式的时间为100ms。

如果引入消息队列，能够大大缩短响应时间，如下：

初识消息中间件之 ==> ActiveMQ

用户注册信息写入数据库后，再将发送邮件和短信写入消息队列，然后直接返回注册结果，总共耗时55m，是并行的一半左右，是串行的三分之一左右，大大提高了系统的处理能力。

应用解耦

场景说明：用户下单后，订单系统需要通知库存系统。传统的做法是，订单系统调用库存系统的接口，如图所示：

初识消息中间件之 ==> ActiveMQ

传统模式的缺点：

假如库存系统无法访问，则订单减库存将失败，从而导致下单失败；
订单系统与库存系统耦合；

引入消息队列的方案如下：

初识消息中间件之 ==> ActiveMQ

订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功。
库存系统：订阅下单的消息，采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作。
假如：在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦。

流量削峰

场景说明：业务系统处理能力远远大于支付渠道处理能力，假如不控制流量把全部请求往支付渠道发送，支付渠道可能会挂掉，导致整个业务不能成功。

这时引入消息队列，控制流量，让请求有序的进入支付渠道

初识消息中间件之 ==> ActiveMQ

日志处理

日志处理是指将消息队列用在日志处理中，比如 Kafka 的应用，解决大量日志传输的问题。架构简化如下：

初识消息中间件之 ==> ActiveMQ

日志采集客户端，负责日志数据采集，定时写受写入Kafka队列；
Kafka消息队列，负责日志数据的接收，存储和转发；
日志处理应用：订阅并消费kafka队列中的日志数据；

三、Active MQ

下载

http://activemq.apache.org/components/classic/download/

安装

直接解压，然后移动到指定目录即可。

>tar zxvf apache-activemq-5.15.10-bin.tar.gz
>mv ./apache-activemq-5.15.10 /usr/local

启动

>/usr/local/activemq-5.15.10/bin/activemq start

# 检查启动状态
[root@cbooy bin]# jps
3168 Jps
2268 activemq.jar

# activemq启动的默认端口号 61616
[root@cbooy bin]# lsof -i:61616
COMMAND  PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java    2268 root  132u  IPv6  15719      0t0  TCP *:61616 (LISTEN)

其他基本命令

> activemq restart # 重启
> activemq stop    # 关闭
> activemq start > /activemq_home/logs/activemq.log   # 落地相关信息,打印日志

指定配置文件的启动

./bin/activemq start xbean:/usr/local/activemq-5.15.10/conf/activemq.xml

后台图形化界面支持

http://127.0.0.1:8161/admin
- 默认用户名/密码, admin/admin
图形化页面相关信息说明
- Number Of Pending Messages
  - 等待消费的消息
  - 未出队列的数量
- Number Of Consumers
  - 消费者数量
- Messages Enqueued
  - 进队消息数,进入队列的总数包括出队的消息数
- Messages Dequeued
  - 出队消息数,即消费者消费后的消息

四、Java操作ActiveMQ

依赖 jar 包

dependencies {
    compile('org.apache.activemq:activemq-all:5.15.9')
    compile('org.apache.activemq:activemq-pool:5.15.9')
}

第一种模式：Queue

生产流程

创建连接工厂对象
从工厂中建立一个连接并开启(Connection)
从连接中建立一个会话(Session)
基于会话建立目的地(Queue)
基于会话创建生产者(Producer)
在会话的基础上创建一条消息(Message)
生产者将消息发出
资源关闭

public class Producer {

  // activemq服务的地址，默认通信端口为61616
  private static final String URL = "tcp://192.168.182.128:61616";

  // 定义队列的名称
  private static final String QUEUE_NAME = "test-Queue";

  public static void main(String[] args) {

    MessageProducer producer = null;
    Session session = null;
    Connection connection = null;

    try {
      // 创建连接工厂对象
      ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);

      // 从工厂中建立一个连接并开启(Connection)
      connection = connectionFactory.createConnection();
      connection.start();

      // 从连接中建立一个会话(Session)
      session = connection.createSession(false, 1);

      // 基于会话建立队列(Queue)
      Queue queue = session.createQueue(QUEUE_NAME);

      // 基于会话创建生产者(Producer)
      producer = session.createProducer(queue);

      for (int i = 0; i < 10; i++) {

        // 在会话的基础上创建一条消息(Message)
        TextMessage textMessage = session.createTextMessage("test-mq：" + i);
        // 生产者将消息发出
        producer.send(textMessage);
      }
    } catch (Exception ex) {
      throw new IllegalStateException(ex);
      // 资源关闭
    } finally {
      try {
        if (null != producer) {
          producer.close();
        }
      } catch (JMSException e) {
        e.printStackTrace();
      }
      try {
        if (null != session) {
          session.close();
        }
      } catch (JMSException e) {
        e.printStackTrace();
      }
      try {
        if (null != connection) {
          connection.close();
        }
      } catch (JMSException e) {
        e.printStackTrace();
      }
    }
  }
}

Producer