RabbitMQ（一）：消息队列MQ

1 消息队列MQ

1.1 MQ简介

1、什么是MQ

MQ全称是Message Queue，直译过来叫做消息队列。MQ本质上是一个队列，遵循先进先出的原则，而队列存放内容为message，主要是作为分布式应用之间实现异步通信的方式。是一种跨进程的通信机制，用于上下游传递消息。

它具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能，成为异步RPC的主要手段之一。现在主流的消息队列有：ActiveMQ、RabbitMQ，Kafka，RocketMQ等。

2、MQ的优势

流量削峰

流量消峰：当流量超过处理上限时，正常情况下，超过上限的流量不做处理。而MQ可以把多出来的一部分流量放入消息队列做缓冲。

举个例子：假设某订单系统每秒最多能处理一万次订单，也就是最多承受的10000qps，正常情况下，下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作，系统是处理不了的，只能限制订单超过一万后不允许用户下单。如果使用消息队列做缓冲，则可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这时有些用户还可以下单但是可能会在十几秒后才能收到下单成功的操作，返回结果会有一段时间的延迟但是仍可以进行下单操作。

应用解耦

以电商应用为例，应用中有订单系统、库存系统、物流系统、支付系统。

耦合：用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。

加入MQ后：当转变成基于消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，中单用户感受不到物流系统的故障，提升系统的可用性。

异常处理

有些服务间调用是异步的，例如 A 调用 B，B 需要花费很长时间执行，但是 A 需要知道 B 什么时候可以执行完，以前一般有两种方式：1）A 过一段时间去调用 B 的查询 api 查询；2）A 提供一个 callback api， B 执行完之后调用 api 通知 A 服务。

以上两种方式都不是很优雅，使用消息队列，可以很方便解决这个问题：A 调用 B 服务后，只需要监听 B 处理完成的消息，当 B 处理完成后，会发送一条消息给 MQ，MQ 会将此消息转发给 A 服务。这样 A 服务既不用循环调用 B 的查询 api，也不用提供 callback api。同样B 服务也不用做这些操作，A 服务还能及时的得到异步处理成功的消息。

数据分发

MQ支持发布订阅机制。支持上下游的一对一、一对多、或者广播模式，并且都可以根据规则选择分发的对象。这样一份上游数据，众多下游系统中，可以根据规则来选择是否接收这些数据。

分布式事务

事务 Transaction：是一种特殊的执行程序，一般具有四个特性：

• a：原子性，要么成功要么失败
• c：一致性，宏观视角下的全局视角一致性
• i：隔离性，并发事务互不相干
• d：持久性，事务提交后，变更永久生效

分布式事务：是指事务的参与者、支持事务的服务器、资源服务器以及事务管理器分别位于不同的分布式系统的不同节点之上。例如在大型电商系统中，下单接口通常会扣减库存、减去优惠、生成订单 id, 而订单服务与库存、优惠、订单 id 都是不同的服务，下单接口的成功与否，不仅取决于本地的 db 操作，而且依赖第三方系统的结果，这时候分布式事务就保证这些操作要么全部成功，要么全部失败。本质上来说，分布式事务就是为了保证不同数据库的数据一致性。
例如：一个用户下单的整个流程，MQ在这其中保证的作用：

生产者的逻辑

• 1、订单入库
• 2、消息记录入库
• 3、发送消息（采用确认模式）
• 4、mq收到消息之后给生产端一个确认消息
• 5、生产端监听这个确认消息
• 6、根据监听结果操作消息表的状态
• 7、定时任务定时去操作消息状态为1未发送的记录，就是那些没有监听到结果的记录进行重新发送

消费者的逻辑

• 1、将收到消息的消息入库
• 2、处理消息失败消息记录的状态就为未处理
• 3、处理消息成功修改消息记录的状态为处理成功
• 4、收到相同的消息id的消息直接丢弃
• 5、定时任务去操作那些未处理，并且已经经过一段时间的消息
• 6、针对那些一直处理失败的，且很长一段时间都没办法处理成功的消息交由人工或者其他途径处理

3、消息中间件的弊端

消息中间件带来诸多好处的同时，也会引入很多的弊端：

• 系统可用性降低：系统可用性在某种程度上降低，比如要考虑消息丢失、消息中间件宕机等问题。
• 系统复杂性提高：引入消息中间件之后，业务需要考虑消息被重复消费、消息丢失、消息传递顺序等问题。
• 一致性问题：消息队列的异步机制确实可以提高系统响应速度，但消费者没有正确消费可能会引入一致性问题。

4、常用的MQ

• ActiveMQ：
  • 优点：单机吞吐量万级，时效性 ms 级，可用性高，基于主从架构实现高可用性
  • 缺点：官方社区现在对 ActiveMQ 5.x 维护越来越少，高吞吐量场景较少使用
• Kafaka：
  • 简单描述：主要应用在大数据领域，有百万级 TPS 的吞吐量，在数据采集、传输、存储的过程中发挥着举足轻重的作用。目前已经被 LinkedIn，Uber, Twitter, Netflix 等大公司所采纳
  • 优点：
    • 性能优越，吞吐量高，单机写入 TPS 约在百万条/秒，时效性 ms 级，可用性非常高
    • 分布式，一个数据多个副本，少数机器宕机，不会丢失数据导致服务不可用
    • 有Web管理界面
    • 日志收集功能成熟，在大数据领域的日志采集中被大规模使用
  • 缺点：
    • Kafka 单机超过 64 个队列/分区，Load 会发生明显的飙高现象，发送消息相应时间变长
    • 消费失败不支持重试
    • 支持消息顺序，一旦有一旦宕机，会造成消息乱序
    • 社区更新较慢
  • 使用场景：Kafka 主要特点是基于Pull 的模式来处理消息消费，追求高吞吐量，适合日志收集和传输。适合产生大量数据的互联网服务的数据收集业务，大型公司建议可以选用，如果需要日志采集功能，首选 kafka 。
• RocketMQ：
  • 简单描述：RocketMQ是阿里的开源产品，使用Java语言实现，被阿里巴巴广泛应用在订单，交易，充值，流计算，消息推送，日志流式处理，binglog 分发等场景。
  • 优点：
    • 单机吞吐量十万级,可用性非常高
    • 分布式架构，扩展性好
    • 消息可以做到 0 丢失，支持 10 亿级别的消息堆积，不会因为堆积导致性能下降
    • MQ功能完善，同时由于的Java开源的，所以可以根据自己公司需求进行定制
  • 缺点：
    • 客户端语言只支持Java和C++，但是C++不成熟
    • 社区活跃度一般
    • 迁移系统时需要修改大量代码
    • 使用场景：多用于金融互联网等对可靠性要求很高的场景，尤其是电商里面的订单扣款，以及业务削峰，在大量交易涌入时，后端可能无法及时处理的情况。
• RabbitMQ
  • 简单描述：2007 年发布，是一个在 AMQP(高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一
  • 优点：
    • 由于 erlang 语言的高并发特性，性能较好，吞吐量到万级
    • 健壮、稳定、易用、跨平台、支持多种语言 如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持 AJAX 文档齐全
    • 社区活跃度高，更新频率高
  • 缺点：商业版需要收费，学习成本高
  • 使用场景：性能好时效性微秒级，社区活跃度也比较高，管理界面用起来十分方便，如果数据量量不大，中小型公司优先选择功能比较完备的 RabbitMQ

1.2 MQ中几个基本概念

队理管理器（queue mr），是MQ系统当中最上层的一个管理器，由它为我们提供基于队列的消息服务

消息（messages）：在MQ中，把应用程序交由MQ传输的数据，某个应用向应用程序发出的请求处理等都可以称做消息。消息又包括两部分：

• 消息描述符（message header,消息头），如： 消息的优先级，生命周期，消息ID
• 消息体（message body）,就是用户的数据部份，（又为分永久性的，和非永久性消息）

队列（queue）：存放消息的地方，队列中的消息直到被应用程序消费之后再结束消息的生命周期。队列又可分为多种类型：

• 本地队列：
  • 普通本地队列：应用程序通过API对其进行读写操作的队列
  • 传输队列：又叫存储转发队列。比如：当某个消息交给MQ系统发送到远程主机，而此时网络发生故障，MQ将把消息放在传输队列中暂存，当网络恢复时，再发往远端目的地
• 远程队列：目的队列在本地的定义，它类似一个地址指针，指向远程主机上的某个目的队列（注意：不真正占用磁盘存储空间）
• 模板队列：是一个模板，当实际需要新增队列时，打开一个模板队列，MQ就会自动生成一个动态队列
• 动态队列：可以指定该动态队列为临时队列或者是永久队列，临时队列可以在关闭的同时进行删除；而永久队列，则可以永久保留
• 别名队列
• 。。。

通道：MQ系统中队列管理器之间传递消息的管道

1.3 MQ的通信模式

MQ可以在内存或者磁盘上，队列中存储消息直到被应用程序读取。应用程序可以通过MQ独立的执行，不需要彼此之间有其他联系，也不需要等待接收程序接收此消息。在分布式计算环境中，为了集成分布式应用，开发者需要对异构网络环境下的分布式应用提供有效的通信手段。为了管理需要共享的信息，对应用提供公共的信息交换机制是重要的。常见的有以下几种：

• 点对点 - 点对点方式是最为传统和常见的通讯方式，它支持一对一、一对多、多对多、多对一等多种配置方式，支持树状、网状等多种拓扑结构。
• 多点广播 - 多点广播可以使用一条 MQ 指令将单一消息发送到多个目标站点，并确保为每一站点可靠地提供信息。
• 发布/订阅 (Publish/Subscribe) - 该模式是将消息按照特定的主题甚至内容进行分发，用户或应用程序可以根据主题或内容接收到所需要的消息。
  • 特点：该模式突破目的队列地理位置的限制。可以让发送者不必关心接收者的目的地址，而接收者也不必关心消息的发送地址，而只是根据消息的主题进行消息的收发。
• 集群 (Cluster) - 集群类似于一个域 (Domain)，集群内部的队列管理器之间通讯时，不需要两两之间建立消息通道，而是采用集群 (Cluster) 通道与其它成员通讯。
  • 特点：简化了系统配置。同时集群中的队列管理器之间能够自动进行负载均衡

1.4 消息的发布策略

之前了解到，MQ是一种常用的异步通信机制。它可以帮助系统的不同组件在无需直接相互通信的情况下，实现数据的传输和处理。消息队列的核心概念之一是消息分发策略（Message Dispatching Strategy），它决定了在发送方发送消息到消息队列后，消息如何被接收方从队列中取出并处理。
消息分发策略是指消息在队列中如何被分配给接收方的规则。常见的消息分发策略有以下几种：

• 先来先服务（First-Come, First-Served，FCFS）：按照消息到达队列的顺序进行分发，先到者先出。
• 最短作业优先（Shortest Job First，SJF）：优先分发队列中处理时间最短的消息。
• 优先级调度（Priority Scheduling）：根据消息的优先级进行分发，优先级高的消息先被处理。
• 轮询调度（Round Robin）：按照顺序轮流分发队列中的消息。
• 随机分发（Random Scheduling）：随机选择队列中的消息进行分发。

1.5 常用消息中间件协议

1、AMQP协议

AMQP协议（Advanced Message Queuing Protocol—高级消息队列协议），一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件的限制。
特点：分布式事务支持、消息的持久化支持、高性能和高可靠的消息处理优势
应用：RabbitMQ、ActiveMQ

2、MQTT协议

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅(publish/subscribe)模式的“轻量级”通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。
特点：以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务；轻量、结构简单、传输快、不支持事务、没有持久化设计
应用：在物联网、小型设备、移动应用等方面有较广泛的应用，RabbitMQ通过插件可以支持该协议

3、OpenMessage协议

OpenMessage协议是由阿里、雅虎和滴滴出行、 Stremalio等公司共同参与创立的分布式消息中间件、流处理等领域的应用开发标准。
特点：结构简单，解析速度快，支持事务和持久化设计

4、kafaka协议

Kafaka协议，基于TCP/IP的二进制协议。消息内部是通过长度来分割，由些基本数据类型组成。
特点：结构简单、解析速度快、无事务支持、有持久化设计