消息队列之RabbitMQ-分布式部署

RabbitMQ分布式部署有3种方式：

Federation与Shovel都是以插件的形式来实现，复杂性相对高，而集群是RabbitMQ的自带属性，相对简单。

这三种方式并不是互斥的，可以根据需求选择相互组合来达到目的。

RabbitMQ本身是基于Erlang编写，Erlang语言天生具备分布式特性（通过同步Erlang集群各节点的magic cookie来实现）。

因此，RabbitMQ天然支持Clustering。这使得RabbitMQ本身不需要像ActiveMQ、Kafka那样通过ZooKeeper分别来实现HA方案和保存集群的元数据。集群是保证可靠性的一种方式，同时可以通过水平扩展以达到增加消息吞吐量能力的目的。

我们把部署RabbitMQ的机器称为节点，也就是broker。broker有2种类型节点： 磁盘节点 和 内存节点 。顾名思义，磁盘节点的broker把元数据存储在磁盘中，内存节点把元数据存储在内存中，很明显，磁盘节点的broker在重启后元数据可以通过读取磁盘进行重建，保证了元数据不丢失，内存节点的broker可以获得更高的性能，但在重启后元数据就都丢了。

元数据包含以下内容：

单节点系统必须是磁盘节点 ，否则每次你重启RabbitMQ之后所有的系统配置信息都会丢失。

集群中至少有一个磁盘节点 ，当节点加入和离开集群时，必须通知磁盘 节点。

如果集群中的唯一一个磁盘节点，结果这个磁盘节点还崩溃了，那会发生什么情况？集群依然可以继续路由消息（因为其他节点元数据在还存在），但无法做以下 *** 作：

也就是说，如果唯一磁盘的磁盘节点崩溃， 集群是可以保持运行的，但不能更改任何东西 。为了增加可靠性，一般会在集群中设置两个磁盘节点，只要任何一个处于工作状态，就可以保障集群的正常服务。

RabbitMQ的集群模式分为两种： 普通模式 与 镜像模式 。

普通模式，也是默认的集群模式。

对于Queue来说， 消息实体只存在于其中一个节点 ，A、B两个节点仅有相同的元数据，即队列结构。当消息进入A节点的Queue中后，consumer从B节点拉取时，RabbitMQ会临时在A、B间进行消息传输，把A中的消息实体取出并经过B发送给consumer。所以consumer应尽量连接每一个节点，从中取消息。即对于同一个逻辑队列，要在多个节点建立物理Queue。否则无论consumer连A或B，出口总在A，会产生瓶颈。

队列所在的节点称为 宿主节点 。

队列创建时，只会在宿主节点创建队列的进程，宿主节点包含完整的队列信息，包括元数据、状态、内容等等。因此， 只有队列的宿主节点才能知道队列的所有信息 。

队列创建后，集群只会同步队列和交换器的元数据到集群中的其他节点，并不会同步队列本身，因此 非宿主节点就只知道队列的元数据和指向该队列宿主节点的指针 。

假如现在一个客户端需要对Queue A进行发布或者订阅，发起与集群的连接，有两种可能的场景：

由于节点之间存在路由转发的情况，对延迟非常敏感，应当只在本地局域网内使用，在广域网中不应该使用集群，而应该用Federation或者Shovel代替。

这样的设计，保证了不论从哪个broker中均可以消费所有队列的数据，并分担了负载，因此，增加broker可以线性提高服务的性能和吞吐量。

但该方案也有显著的缺陷，那就是 不能保证消息不会丢失 。当集群中某一节点崩溃时，崩溃节点所在的队列进程和关联的绑定都会消失，附加在那些队列上的消费者也会丢失其订阅信息，匹配该队列的新消息也会丢失。比如A为宿主节点，当A节点故障后，B节点无法取到A节点中还未消费的消息实体。如果做了消息持久化，那么得等A节点恢复，然后才可被消费；如果没有持久化的话，然后就没有然后了……

肯定有不少同学会问，想要实现HA方案，那将RabbitMQ集群中的所有Queue的完整数据在所有节点上都保存一份不就可以了吗？比如类似MySQL的主主模式，任何一个节点出现故障或者宕机不可用时，那么使用者的客户端只要能连接至其他节点，不就能够照常完成消息的发布和订阅吗？

RabbitMQ这么设计是基于性能和存储空间上来考虑：

引入 镜像队列 （Mirror Queue）的机制，可以将队列镜像到集群中的其他Broker节点之上，如果集群中的一个节点失效了，队列能够自动切换到镜像中的另一个节点上以保证服务的可用性。

一个镜像队列中包含有1个主节点master和若干个从节点slave。其主从节点包含如下几个特点：

该模式和普通模式不同之处在于，消息实体会主动在镜像节点间同步，而不是在consumer取数据时临时拉取。该模式带来的副作用也很明显，除了降低系统性能外，如果镜像队列数量过多，加之大量的消息进入，集群内部的网络带宽将会被这种同步通讯大大消耗掉。所以在对可靠性要求较高的场合中适用。

一个队列想做成镜像队列，需要先设置policy，然后客户端创建队列的时候，rabbitmq集群根据队列名称自动设置为普通队列还是镜像队列。

镜像队列的配置通过添加policy完成，policy添加的命令为：

例如，对队列名称以hello开头的所有队列进行镜像，并在集群的两个节点上完成镜像，policy的设置命令为：

rabbitmqctl set_policy hello-ha "^hello" '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

通常队列由两部分组成：一部分是AMQQueue，负责AMQP协议相关的消息处理，即接收生产者发布的消息、向消费者投递消息、处理消息confirm、acknowledge等等；另一部分是BackingQueue，它提供了相关的接口供AMQQueue调用，完成消息的存储以及可能的持久化工作等。

镜像队列基本上就是一个特殊的BackingQueue，它内部包裹了一个普通的BackingQueue做本地消息持久化处理，在此基础上增加了将消息和ack复制到所有镜像的功能。所有对mirror_queue_master的 *** 作，会通过组播GM（下面会讲到）的方式同步到各slave节点。GM负责消息的广播，mirror_queue_slave负责回调处理，而master上的回调处理是由coordinator负责完成。mirror_queue_slave中包含了普通的BackingQueue进行消息的存储，master节点中BackingQueue包含在mirror_queue_master中由AMQQueue进行调用。

消息的发布（除了BasicPublish之外）与消费都是通过master节点完成。master节点对消息进行处理的同时将消息的处理动作通过GM广播给所有的slave节点，slave节点的GM收到消息后，通过回调交由mirror_queue_slave进行实际的处理。

GM(Guarenteed Multicast) 是一种可靠的组播通讯协议，该协议能够保证组播消息的原子性，即保证组中活着的节点要么都收到消息要么都收不到。它的实现大致如下：

将所有的节点形成一个循环链表，每个节点都会监控位于自己左右两边的节点，当有节点新增时，相邻的节点保证当前广播的消息会复制到新的节点上；当有节点失效时，相邻的节点会接管保证本次广播的消息会复制到所有的节点。在master节点和slave节点上的这些gm形成一个group，group（gm_group）的信息会记录在mnesia中。不同的镜像队列形成不同的group。消息从master节点对于的gm发出后，顺着链表依次传送到所有的节点，由于所有节点组成一个循环链表，master节点对应的gm最终会收到自己发送的消息，这个时候master节点就知道消息已经复制到所有的slave节点了。

slave节点先从gm_group中获取对应group的所有成员信息，然后随机选择一个节点并向这个节点发送请求，这个节点收到请求后，更新gm_group对应的信息，同时通知左右节点更新邻居信息（调整对左右节点的监控）及当前正在广播的消息，然后回复通知请求节点成功加入group。请求加入group的节点收到回复后再更新rabbit_queue中的相关信息，并根据需要进行消息的同步。

当slave节点失效时，仅仅是相邻节点感知，然后重新调整邻居节点信息、更新rabbit_queue、gm_group的记录等。如果是master节点失效，"资格最老"的slave节点被提升为master节点，slave节点会创建出新的coordinator，并告知gm修改回调处理为coordinator，原来的mirror_queue_slave充当amqqueue_process处理生产者发布的消息，向消费者投递消息等。

上面提到如果是slave节点失效，只有相邻的节点能感知到，那么master节点失效是不是也是只有相邻的节点能感知到？假如是这样的话，如果相邻的节点不是"资格最老"的节点，怎么通知"资格最老"的节点提升为新的master节点呢

实际上，所有的slave节点在加入group时，mirror_queue_slave进程会对master节点的amqqueue_process进程（也可能是mirror_queue_slave进程）进行监控，如果master节点失效的话，mirror_queue_slave会感知，然后再通过gm进行广播，这样所有的节点最终都会知道master节点失效。当然，只有"资格最老"的节点会提升自己为新的master。

消息从master节点发出，顺着节点链表发送。在这期间，所有的slave节点都会对消息进行缓存，当master节点收到自己发送的消息后，会再次广播ack消息，同样ack消息会顺着节点链表经过所有的slave节点，其作用是通知slave节点可以清除缓存的消息，当ack消息回到master节点时对应广播消息的生命周期结束。

下图为一个简单的示意图，A节点为master节点，广播一条内容为"test"的消息。"1"表示消息为广播的第一条消息；"id=A"表示消息的发送者为节点A。右边是slave节点记录的状态信息。

为什么所有的节点都需要缓存一份发布的消息呢？

master发布的消息是依次经过所有slave节点，在这期间的任何时刻，有可能有节点失效，那么相邻的节点可能需要重新发送给新的节点。例如，A->B->C->D->A形成的循环链表，A为master节点，广播消息发送给节点B，B再发送给C，如果节点C收到B发送的消息还未发送给D时异常结束了，那么节点B感知后节点C失效后需要重新将消息发送给D。同样，如果B节点将消息发送给C后，B,C节点中新增了E节点，那么B节点需要再将消息发送给新增的E节点。

配置镜像队列的时候有个 ha-sync-mode 属性，这个有什么用呢

新节点加入到group后，最多能从左边节点获取到当前正在广播的消息内容，加入group之前已经广播的消息则无法获取到。如果此时master节点不幸失效，而新节点有恰好成为了新的master，那么加入group之前已经广播的消息则会全部丢失。

注意：这里的消息具体是指新节点加入前已经发布并复制到所有slave节点的消息，并且这些消息还未被消费者消费或者未被消费者确认。如果新节点加入前，所有广播的消息被消费者消费并确认了，master节点删除消息的同时会通知slave节点完成相应动作。这种情况等同于新节点加入前没有发布任何消息。

避免这种问题的解决办法就是对新的slave节点进行消息同步。当 ha-sync-mode 配置为自动同步（automatic）时，新节点加入group时会自动进行消息的同步；如果配置为manually则需要手动 *** 作完成同步。

Federation直译过来是联邦，它的设计目标是使 RabbitMQ 在不同的 Broker 节点之间进行消息传递而无须建

立集群。具有以下特点：

那么它到底有什么用呢？我们可以从一个实际场景入手：

有两个服务分别部署在国内和海外，它们之间需要通过消息队列来通讯。

很明显无论RabbitMQ部署在海外还是国内，另一方一定得忍受连接上的延迟。因此我们可以在海外和国内各部署一个MQ，这样一来海外连接海外的MQ,国内连接国内，就不会有连接上的延迟了。

但这样还会有问题，假设某生产者将消息存入海外MQ中的某个队列 queueB ， 在国内的服务想要消费 queueB 消息，消息的流转及确认必然要忍受较大的网络延迟 ，内部编码逻辑也会因这一因素变得更加复杂。

此外，服务可能得维护两个MQ的配置，比如国内服务在生产消息时得使用国内MQ，消费消息时得监听海外MQ的队列，降低了系统的维护性。

可能有人想到可以用集群，但是RabbitMQ的集群对延迟非常敏感，一般部署在局域网内，如果部署在广域网可能会产生网络分区等等问题。

这时候，Federation就派上用场了。它被设计成能够容忍不稳定的网络连接情况，完全能够满足这样的场景。

那使用Federation之后是怎样的业务流程呢

首先我们在海外MQ上定义exchangeA，它通过路由键“rkA”绑定着queueA。然后用Federation在exchangeA上建立一条 单向 连接到国内RabbitMQ，Federation则自动会在国内RabbitMQ建立一个exchangeA交换器（默认同名）。

这时候，如果部署在国内的client C在国内MQ上publish了一条消息，这条消息会通过 Federation link 转发到海外MQ的交换器exchangeA中，最终消息会存入与 exchangeA 绑定的队列 queueA 中，而client C也能立即得到返回。

实际上，Federation插件还会在国内MQ建立一个内部的交换器：exchangeA→ broker3 B（broker3是集群名），并通过路由键 "rkA"将它和国内MQ的exchangeA绑定起来。接下来还会在国内MQ上建立一个内部队列federation: exchangeA->broker3 B，并与内部exchange绑定。这些 *** 作都是内部的，对客户端来说是透明的。

值得一提的是，Federation的连接是单向的，如果是在海外MQ的exchangeA上发送消息是不会转到国内的。

这种在exchange上建立连接进行联邦的，就叫做 联邦交换器 。一个联邦交换器接收上游（upstream）的信息，这里的上游指的是其他的MQ节点。

对比前面举的例子，国内MQ就是上游，联邦交换器能够将原本发送给上游交换器的消息路由到本地的某个队列中。

有联邦交换器自然也有联播队列，联邦队列则允许一个本地消费者接收到来自上游队列的消息 。

如图，海外MQ有队列A，给其设置一条链接，Federation则自动会在国内RabbitMQ建立一个队列A（默认同名）。

当有消费者 ClinetA连接海外MQ并消费 queueA 中的消息时，如果队列 queueA中本身有若干消息堆积，那么 ClientA直接消费这些消息，此时海外MQ中的queueA并不会拉取国内中的 queueA 的消息；如果队列 queueA中没有消息堆积或者消息被消费完了，那么它会通过 Federation link 拉取上游队列 queueA 中的消息(如果有消息)，然后存储到本地，之后再被消费者 ClientA进行消费 。

首先开启Federation 功能：

值得注意的是，当需要在集群中使用 Federation 功能的时候，集群中所有的节点都应该开启 Federation 插件。

接下来我们要配置两个东西：upstreams和Policies。

每个 upstream 用于定义与其他 Broker 建立连接的信息。

通用参数如下：

然后定义一个 Policy， 用于匹配交换器：

^exchange 意思是将匹配所有以exchange名字开头的交换器，为它们在上游创建连接。这样就创建了一个 Federation link。

Shovel是RabbitMQ的一个插件， 能够可靠、持续地从一个Broker 中的队列（作为源端，即source ）拉取数据并转发至另一个Broker 中的交换器（作为目的端，即destination ）。作为源端的队列和作为目的端的交换器可以同时位于同一个 Broker 上，也可以位于不同的 Broker 上。

使用Shovel有以下优势：

使用Shovel时，通常源为队列，目的为交换器：

但是，也可以源为队列，目的为队列。实际也是由交换器转发，只不过这个交换器是默认交换器。配置交换器做为源也是可行的。实际上会在源端自动新建一个队列，消息先存在这个队列，再被Shovel移走。

使用Shovel插件命令：

Shovel 既可以部署在源端，也可以部署在目的端。有两种方式可以部署 Shovel：

其主要差异如下：

来看一个使用Shovel治理消息堆积的案例。

当某个队列中的消息堆积严重时，比如超过某个设定的阈值，就可以通过 Shovel 将队列中的消息移交给另一个集群。

我们的服务器从单机发展到拥有多台机器的分布式系统，各个系统之前需要借助于网络进行通信，原有单机中相对可靠的方法调用以及进程间通信方式已经没有办法使用，同时网络环境也是不稳定的，造成了我们多个机器之间的数据同步问题，这就是典型的分布式事务问题。

在分布式事务中事务的参与者、支持事务的服务器、资源服务器以及事务管理器分别位于不同的分布式系统的不同节点之上。分布式事务就是要保证不同节点之间的数据一致性。

1、2PC(二阶段提交)方案 - 强一致性

2、3PC(三阶段提交)方案

3、TCC （Try-Confirm-Cancel）事务 - 最终一致性

4、Saga事务 - 最终一致性

5、本地消息表 - 最终一致性

6、MQ事务 - 最终一致性

消息的生产方，除了维护自己的业务逻辑之外，同时需要维护一个消息表。这个消息表里面记录的就是需要同步到别的服务的信息，当然这个消息表，每个消息都有一个状态值，来标识这个消息有没有被成功处理。

发送放的业务逻辑以及消息表中数据的插入将在一个事务中完成，这样避免了业务处理成功 + 事务消息发送失败，或业务处理失败 + 事务消息发送成功，这个问题。

举个栗子：

我们假定目前有两个服务，订单服务，购物车服务，用户在购物车中对几个商品进行合并下单，之后需要情况购物车中刚刚已经下单的商品信息。

1、消息的生产方也就是订单服务，完成了自己的逻辑(对商品进行下单 *** 作)然后把这个消息通过 mq 发送到需要进行数据同步的其他服务中，也就是我们栗子中的购物车服务。

2、其他服务(购物车服务)会监听这个队列；

1、如果收到这个消息，并且数据同步执行成功了，当然这也是一个本地事务，就通过 mq 回复消息的生产方(订单服务)消息已经处理了，然后生产方就能标识本次事务已经结束。如果是一个业务上的错误,就回复消息的生产方，需要进行数据回滚了。

2、很久没收到这个消息，这种情况是不会发生的，消息的发送方会有一个定时的任务，会定时重试发送消息表中还没有处理的消息；

3、消息的生产方(订单服务)如果收到消息回执；

1、成功的话就修改本次消息已经处理完，也就是本次分布式事务的同步已经完成；

2、如果消息的结果是执行失败，同时在本地回滚本次事务，标识消息已经处理完成；

3、如果消息丢失，也就是回执消息没有收到，这种情况也不太会发生，消息的发送方(订单服务)会有一个定时的任务，定时重试发送消息表中还没有处理的消息，下游的服务需要做幂等，可能会收到多次重复的消息，如果一个回复消息生产方中的某个回执信息丢失了，后面持续收到生产方的 mq 消息，然后再次回复消息的生产方回执信息，这样总能保证发送者能成功收到回执，消息的生产方在接收回执消息的时候也要做到幂等性。

这里有两个很重要的 *** 作：

1、服务器处理消息需要是幂等的，消息的生产方和接收方都需要做到幂等性；

2、发送放需要添加一个定时器来遍历重推未处理的消息，避免消息丢失，造成的事务执行断裂。

该方案的优缺点

优点：

1、在设计层面上实现了消息数据的可靠性，不依赖消息中间件，弱化了对 mq 特性的依赖。

2、简单，易于实现。

缺点：

主要是需要和业务数据绑定到一起，耦合性比较高，使用相同的数据库，会占用业务数据库的一些资源。

下面分析下几种消息队列对事务的支持

RocketMQ 中的事务，它解决的问题是，确保执行本地事务和发消息这两个 *** 作，要么都成功，要么都失败。并且，RocketMQ 增加了一个事务反查的机制，来尽量提高事务执行的成功率和数据一致性。

主要是两个方面，正常的事务提交和事务消息补偿

正常的事务提交

1、发送消息（half消息），这个 half 消息和普通消息的区别，在事务提交 之前，对于消费者来说，这个消息是不可见的。

2、MQ SERVER写入信息，并且返回响应的结果；

3、根据MQ SERVER响应的结果，决定是否执行本地事务,如果MQ SERVER写入信息成功执行本地事务，否则不执行；

如果MQ SERVER没有收到 Commit 或者 Rollback 的消息，这种情况就需要进行补偿流程了

补偿流程

1、MQ SERVER如果没有收到来自消息发送方的 Commit 或者 Rollback 消息，就会向消息发送端也就是我们的服务器发起一次查询，查询当前消息的状态；

2、消息发送方收到对应的查询请求，查询事务的状态，然后把状态重新推送给MQ SERVER，MQ SERVER就能之后后续的流程了。

相比于本地消息表来处理分布式事务，MQ 事务是把原本应该在本地消息表中处理的逻辑放到了 MQ 中来完成。

Kafka 中的事务解决问题，确保在一个事务中发送的多条信息，要么都成功，要么都失败。也就是保证对多个分区写入 *** 作的原子性。

通过配合 Kafka 的幂等机制来实现 Kafka 的 Exactly Once，满足了读取-处理-写入这种模式的应用程序。当然 Kafka 中的事务主要也是来处理这种模式的。

什么是读取-处理-写入模式呢？

栗如：在流计算中，用 Kafka 作为数据源，并且将计算结果保存到 Kafka 这种场景下，数据从 Kafka 的某个主题中消费，在计算集群中计算，再把计算结果保存在 Kafka 的其他主题中。这个过程中，要保证每条消息只被处理一次，这样才能保证最终结果的成功。Kafka 事务的原子性就保证了，读取和写入的原子性，两者要不一起成功，要不就一起失败回滚。

这里来分析下 Kafka 的事务是如何实现的

它的实现原理和 RocketMQ 的事务是差不多的，都是基于两阶段提交来实现的，在实现上可能更麻烦

先来介绍下事务协调者，为了解决分布式事务问题，Kafka 引入了事务协调者这个角色，负责在服务端协调整个事务。这个协调者并不是一个独立的进程，而是 Broker 进程的一部分，协调者和分区一样通过选举来保证自身的可用性。

Kafka 集群中也有一个特殊的用于记录事务日志的主题，里面记录的都是事务的日志。同时会有多个协调者的存在，每个协调者负责管理和使用事务日志中的几个分区。这样能够并行的执行事务，提高性能。

下面看下具体的流程

事务的提交

1、协调者设置事务的状态为PrepareCommit，写入到事务日志中；

2、协调者在每个分区中写入事务结束的标识，然后客户端就能把之前过滤的未提交的事务消息放行给消费端进行消费了；

事务的回滚

1、协调者设置事务的状态为PrepareAbort，写入到事务日志中；

2、协调者在每个分区中写入事务回滚的标识，然后之前未提交的事务消息就能被丢弃了；

这里引用一下消息队列高手课中的

RabbitMQ 中事务解决的问题是确保生产者的消息到达MQ SERVER，这和其他 MQ 事务还是有点差别的，这里也不展开讨论了。

先来分析下一条消息在 MQ 中流转所经历的阶段。

生产阶段 ：生产者产生消息，通过网络发送到 Broker 端。

存储阶段 ：Broker 拿到消息，需要进行落盘，如果是集群版的 MQ 还需要同步数据到其他节点。

消费阶段 ：消费者在 Broker 端拉数据，通过网络传输到达消费者端。

发生网络丢包、网络故障等这些会导致消息的丢失

在生产者发送消息之前，通过channeltxSelect开启一个事务，接着发送消息， 如果消息投递 server 失败，进行事务回滚channeltxRollback，然后重新发送， 如果 server 收到消息，就提交事务channeltxCommit

不过使用事务性能不好，这是同步 *** 作，一条消息发送之后会使发送端阻塞，以等待RabbitMQ Server的回应，之后才能继续发送下一条消息，生产者生产消息的吞吐量和性能都会大大降低。

使用确认机制，生产者将信道设置成 confirm 确认模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都会被指派一个唯一的ID（从1开始），一旦消息被投递到所有匹配的队列之后，RabbitMQ 就会发送一个确认（BasicAck）给生产者（包含消息的唯一 deliveryTag 和 multiple 参数），这就使得生产者知晓消息已经正确到达了目的地了。

multiple 为 true 表示的是批量的消息确认，为 true 的时候，表示小于等于返回的 deliveryTag 的消息 id 都已经确认了，为 false 表示的是消息 id 为返回的 deliveryTag 的消息，已经确认了。

确认机制有三种类型

1、同步确认

2、批量确认

3、异步确认

同步模式的效率很低，因为每一条消息度都需要等待确认好之后，才能处理下一条；

批量确认模式相比同步模式效率是很高，不过有个致命的缺陷，一旦回复确认失败，当前确认批次的消息会全部重新发送，导致消息重复发送；

异步模式就是个很好的选择了，不会有同步模式的阻塞问题，同时效率也很高，是个不错的选择。

Kafaka 中引入了一个 broker。 broker 会对生产者和消费者进行消息的确认，生产者发送消息到 broker，如果没有收到 broker 的确认就可以选择继续发送。

只要 Producer 收到了 Broker 的确认响应，就可以保证消息在生产阶段不会丢失。有些消息队列在长时间没收到发送确认响应后，会自动重试，如果重试再失败，就会以返回值或者异常的方式告知用户。

只要正确处理 Broker 的确认响应，就可以避免消息的丢失。

RocketMQ 提供了3种发送消息方式，分别是：

同步发送：Producer 向 broker 发送消息，阻塞当前线程等待 broker 响应 发送结果。

异步发送：Producer 首先构建一个向 broker 发送消息的任务，把该任务提交给线程池，等执行完该任务时，回调用户自定义的回调函数，执行处理结果。

Oneway发送：Oneway 方式只负责发送请求，不等待应答，Producer 只负责把请求发出去，而不处理响应结果。

在存储阶段正常情况下，只要 Broker 在正常运行，就不会出现丢失消息的问题，但是如果 Broker 出现了故障，比如进程死掉了或者服务器宕机了，还是可能会丢失消息的。

防止在存储阶段消息额丢失，可以做持久化，防止异常情况(重启，关闭，宕机)。。。

RabbitMQ 持久化中有三部分：

消息的持久化，在投递时指定 delivery_mode=2（1是非持久化），消息的持久化，需要配合队列的持久，只设置消息的持久化，重启之后队列消失，继而消息也会丢失。所以如果只设置消息持久化而不设置队列的持久化意义不大。

对于持久化，如果所有的消息都设置持久化，会影响写入的性能，所以可以选择对可靠性要求比较高的消息进行持久化处理。

不过消息持久化并不能百分之百避免消息的丢失

比如数据在落盘的过程中宕机了，消息还没及时同步到内存中，这也是会丢数据的，这种问题可以通过引入镜像队列来解决。

镜像队列的作用：引入镜像队列，可已将队列镜像到集群中的其他 Broker 节点之上，如果集群中的一个节点失效了，队列能够自动切换到镜像中的另一个节点上来保证服务的可用性。(更细节的这里不展开讨论了)

*** 作系统本身有一层缓存，叫做 Page Cache，当往磁盘文件写入的时候，系统会先将数据流写入缓存中。

Kafka 收到消息后也会先存储在也缓存中(Page Cache)中，之后由 *** 作系统根据自己的策略进行刷盘或者通过 fsync 命令强制刷盘。如果系统挂掉，在 PageCache 中的数据就会丢失。也就是对应的 Broker 中的数据就会丢失了。

处理思路

1、控制竞选分区 leader 的 Broker。如果一个 Broker 落后原先的 Leader 太多，那么它一旦成为新的 Leader，必然会造成消息的丢失。

2、控制消息能够被写入到多个副本中才能提交，这样避免上面的问题1。

1、将刷盘方式改成同步刷盘；

2、对于多个节点的 Broker，需要将 Broker 集群配置成：至少将消息发送到 2 个以上的节点，再给客户端回复发送确认响应。这样当某个 Broker 宕机时，其他的 Broker 可以替代宕机的 Broker，也不会发生消息丢失。

消费阶段就很简单了，如果在网络传输中丢失，这个消息之后还会持续的推送给消费者，在消费阶段我们只需要控制在业务逻辑处理完成之后再去进行消费确认就行了。

总结：对于消息的丢失，也可以借助于本地消息表的思路，消息产生的时候进行消息的落盘，长时间未处理的消息，使用定时重推到队列中。

消息在 MQ 中的传递，大致可以归类为下面三种：

1、At most once: 至多一次。消息在传递时，最多会被送达一次。是不安全的，可能会丢数据。

2、At least once: 至少一次。消息在传递时，至少会被送达一次。也就是说，不允许丢消息，但是允许有少量重复消息出现。

3、Exactly once：恰好一次。消息在传递时，只会被送达一次，不允许丢失也不允许重复，这个是最高的等级。

大部分消息队列满足的都是At least once，也就是可以允许重复的消息出现。

我们消费者需要满足幂等性,通常有下面几种处理方案

1、利用数据库的唯一性

根据业务情况，选定业务中能够判定唯一的值作为数据库的唯一键，新建一个流水表，然后执行业务 *** 作和流水表数据的插入放在同一事务中，如果流水表数据已经存在，那么就执行失败，借此保证幂等性。也可先查询流水表的数据，没有数据然后执行业务，插入流水表数据。不过需要注意，数据库读写延迟的情况。

2、数据库的更新增加前置条件

3、给消息带上唯一ID

每条消息加上唯一ID,利用方法1中通过增加流水表，借助数据库的唯一性来处理重复消息的消费。

前言

开源社区有好多优秀的队列中间件，比如RabbitMQ和Kafka，每个队列都貌似有其特性，在进行工程选择时，往往眼花缭乱，不知所措。对于RabbitMQ和Kafka，到底应该选哪个？

RabbitMQ架构

概念

RabbitMQ是一个分布式系统

broker：每个节点运行的服务程序，功能为维护该节点的队列的增删以及转发队列 *** 作请求。

master queue：每个队列都分为一个主队列和若干个镜像队列。

mirror queue：镜像队列，作为master queue的备份。在master queue所在节点挂掉之后，系统把mirror queue提升为master queue，负责处理客户端队列 *** 作请求。注意，mirror queue只做镜像，设计目的不是为了承担客户端读写压力。

如上图所示，集群中有两个节点，每个节点上有一个broker，每个broker负责本机上队列的维护，并且borker之间可以互相通信。集群中有两个队列A和B，每个队列都分为master queue和mirror queue（备份）。那么队列上的生产消费怎么实现的呢？

队列消费

如上图有两个consumer消费队列A，这两个consumer连在了集群的不同机器上。RabbitMQ集群中的任何一个节点都拥有集群上所有队列的元信息，所以连接到集群中的任何一个节点都可以，主要区别在于有的consumer连在master queue所在节点，有的连在非master queue节点上。

因为mirror queue要和master queue保持一致，故需要同步机制，正因为一致性的限制，导致所有的读写 *** 作都必须都 *** 作在master queue上（想想，为啥读也要从master queue中读？和数据库读写分离是不一样的。），然后由master节点同步 *** 作到mirror queue所在的节点。即使consumer连接到了非master queue节点，该consumer的 *** 作也会被路由到master queue所在的节点上，这样才能进行消费。

队列生产

原理和消费一样，如果连接到非 master queue 节点，则路由过去。

不足

由于master queue单节点，导致性能瓶颈，吞吐量受限。虽然为了提高性能，内部使用了Erlang这个语言实现，但是终究摆脱不了架构设计上的致命缺陷。

Kafka

说实话，Kafka我觉得就是看到了RabbitMQ这个缺陷才设计出的一个改进版，改进的点就是：把一个队列的单一master变成多个master，即一台机器扛不住qps，那么我就用多台机器扛qps，把一个队列的流量均匀分散在多台机器上不就可以了么？注意，多个master之间的数据没有交集，即一条消息要么发送到这个master queue，要么发送到另外一个master queue。

这里面的每个master queue 在Kafka中叫做Partition，即一个分片。一个队列有多个主分片，每个主分片又有若干副分片做备份，同步机制类似于RabbitMQ。

如上图，我们省略了不同的queue，假设集群上只有一个queue（Kafka中叫Topic）。每个生产者随机把消息发送到主分片上，之后主分片再同步给副分片。

队列读取的时候虚拟出一个Group的概念，一个Topic内部的消息，只会路由到同Group内的一个consumer上，同一个Group中的consumer消费的消息是不一样的；Group之间共享一个Topic，看起来就是一个队列的多个拷贝。所以，为了达到多个Group共享一个Topic数据，Kafka并不会像RabbitMQ那样消息消费完毕立马删除，而是必须在后台配置保存日期，即只保存最近一段时间的消息，超过这个时间的消息就会从磁盘删除，这样就保证了在一个时间段内，Topic数据对所有Group可见（这个特性使得Kafka非常适合做一个公司的数据总线）。队列读同样是读主分片，并且为了优化性能，消费者与主分片有一一的对应关系，如果消费者数目大于分片数，则存在某些消费者得不到消息。

由此可见，Kafka绝对是为了高吞吐量设计的，比如设置分片数为100，那么就有100台机器去扛一个Topic的流量，当然比RabbitMQ的单机性能好。

总结

本文只做了Kafka和RabbitMQ的对比，但是开源队列岂止这两个，ZeroMQ，RocketMQ，JMQ等等，时间有限也就没有细看，故不在本文比较范围之内。

所以，别再被这些五花八门的队列迷惑了，从架构上找出关键差别，并结合自己的实际需求（比如本文就只单单从吞吐量一个需求来考察）轻轻松松搞定选型。最后总结如下：

吞吐量较低：Kafka和RabbitMQ都可以。

吞吐量高：Kafka。

以上就是关于消息队列之RabbitMQ-分布式部署全部的内容，包括:消息队列之RabbitMQ-分布式部署、RabbitMQ，RocketMQ，Kafka 事务性，消息丢失和重复发送处理策略、RabbitMQ和Kafka对比，总结了以下几个点等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！