前端工程师认识RabbitMQ的过程

2023-05-08 13:40:03物联网0167

前端工程师认识RabbitMQ的过程,第1张

消息队列（Message Queue）是一种进程间通信或同一进程的不同线程间的通信方式。

Broker（消息服务器）

Broker的概念来自与Apache ActiveMQ，通俗的讲就是MQ的服务器。

Producer（生产者）

业务的发起方，负责生产消息传输给broker

Consumer（消费者）

业务的处理方，负责从broker获取消息并进行业务逻辑处理

Topic（主题）

发布订阅模式下的消息统一汇集地，不同生产者向topic发送消息，由MQ服务器分发到不同的订阅者，实现消息的广播

Queue（队列）

PTP模式下，特定生产者向特定queue发送消息，消费者订阅特定的queue完成指定消息的接收。

Message（消息体）

根据不同通信协议定义的固定格式进行编码的数据包，来封装业务数据，实现消息的传输

点对点模型用于消息生产者和消息消费者之间点到点的通信。

点对点模式包含三个角色：

每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，可以放在内存中也可以持久化，直到他们被消费或超时。

特点：

发布订阅模型包含三个角色：

多个发布者将消息发送到Topic，系统将这些消息传递给多个订阅者。

特点：

AMQP即Advanced Message Queuing Protocol，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

优点：可靠、通用

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的一个即时通讯协议，有可能成为物联网的重要组成部分。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和致动器（比如通过Twitter让房屋联网）的通信协议。

优点：格式简洁、占用带宽小、移动端通信、PUSH、嵌入式系统

STOMP（Streaming Text Orientated Message Protocol）是流文本定向消息协议，是一种为MOM(Message Oriented Middleware，面向消息的中间件)设计的简单文本协议。STOMP提供一个可互操作的连接格式，允许客户端与任意STOMP消息代理（Broker）进行交互。

优点：命令模式（非topic\queue模式）

XMPP（可扩展消息处理现场协议，Extensible Messaging and Presence Protocol）是基于可扩展标记语言（XML）的协议，多用于即时消息（IM）以及在线现场探测。适用于服务器之间的准即时操作。核心是基于XML流传输，这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息，即使其操作系统和浏览器不同。

优点：通用公开、兼容性强、可扩展、安全性高，但XML编码格式占用带宽大

RabbitMQ 是实现 AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。 RabbitMQ 主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

RabbitMQ 是一个开源的 AMQP 实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

Channel（通道）

道是两个管理器之间的一种单向点对点的的通信连接，如果需要双向交流，可以建立一对通道。

Exchange（消息交换机）

Exchange类似于数据通信网络中的交换机，提供消息路由策略。

RabbitMq中，producer不是通过信道直接将消息发送给queue，而是先发送给Exchange。一个Exchange可以和多个Queue进行绑定，producer在传递消息的时候，会传递一个ROUTING_KEY，Exchange会根据这个ROUTING_KEY按照特定的路由算法，将消息路由给指定的queue。和Queue一样，Exchange也可设置为持久化，临时或者自动删除。

Exchange有4种类型：direct(默认)，fanout， topic，和headers。

不同类型的Exchange转发消息的策略有所区别：

Binding（绑定）

所谓绑定就是将一个特定的 Exchange 和一个特定的 Queue 绑定起来。Exchange 和Queue的绑定可以是多对多的关系。

Routing Key（路由关键字）

exchange根据这个关键字进行消息投递。

vhost（虚拟主机）

在RabbitMq server上可以创建多个虚拟的message broker，又叫做virtual hosts (vhosts)。每一个vhost本质上是一个mini-rabbitmq server，分别管理各自的exchange，和bindings。vhost相当于物理的server，可以为不同app提供边界隔离，使得应用安全的运行在不同的vhost实例上，相互之间不会干扰。producer和consumer连接rabbit server需要指定一个vhost。

假设P1和C1注册了相同的Broker，Exchange和Queue。P1发送的消息最终会被C1消费。

基本的通信流程大概如下所示：

Consumer收到消息时需要显式的向rabbit broker发送basic。ack消息或者consumer订阅消息时设置auto_ack参数为true。

在通信过程中，队列对ACK的处理有以下几种情况：

即消息的Ackownledge确认机制，为了保证消息不丢失，消息队列提供了消息Acknowledge机制，即ACK机制，当Consumer确认消息已经被消费处理，发送一个ACK给消息队列，此时消息队列便可以删除这个消息了。如果Consumer宕机/关闭，没有发送ACK，消息队列将认为这个消息没有被处理，会将这个消息重新发送给其他的Consumer重新消费处理。

消息的收发处理支持事务，例如：在任务中心场景中，一次处理可能涉及多个消息的接收、处理，这应该处于同一个事务范围内，如果一个消息处理失败，事务回滚，消息重新回到队列中。

消息的持久化，对于一些关键的核心业务来说是非常重要的，启用消息持久化后，消息队列宕机重启后，消息可以从持久化存储恢复，消息不丢失，可以继续消费处理。

fanout 模式

模式特点：

direct 模式

任何发送到Direct Exchange的消息都会被转发到routing_key中指定的Queue。

如果一个exchange 声明为direct，并且bind中指定了routing_key，那么发送消息时需要同时指明该exchange和routing_key。

简而言之就是：生产者生成消息发送给Exchange， Exchange根据Exchange类型和basic_publish中的routing_key进行消息发送消费者：订阅Exchange并根据Exchange类型和binding key(bindings 中的routing key) ，如果生产者和订阅者的routing_key相同，Exchange就会路由到那个队列。

topic 模式

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。

topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同。

它约定：

以上图中的配置为例，routingKey=”quickorangerabbit”的消息会同时路由到Q1与Q2，routingKey=”lazyorangefox”的消息会路由到Q1，routingKey=”lazybrownfox”的消息会路由到Q2，routingKey=”lazypinkrabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quickbrownfox”、routingKey=”orange”、routingKey=”quickorangemalerabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。

RabbitMQ，部署分三种模式：单机模式，普通集群模式，镜像集群模式。

普通集群模式

多台机器部署，每个机器放一个rabbitmq实例，但是创建的queue只会放在一个rabbitmq实例上，每个实例同步queue的元数据。

如果消费时连的是其他实例，那个实例会从queue所在实例拉取数据。这就会导致拉取数据的开销，如果那个放queue的实例宕机了，那么其他实例就无法从那个实例拉取，即便开启了消息持久化，让rabbitmq落地存储消息的话，消息不一定会丢，但得等这个实例恢复了，然后才可以继续从这个queue拉取数据， 这就没什么高可用可言，主要是提供吞吐量 ，让集群中多个节点来服务某个queue的读写操作。

镜像集群模式

queue的元数据和消息都会存放在多个实例，每次写消息就自动同步到多个queue实例里。这样任何一个机器宕机，其他机器都可以顶上，但是性能开销太大，消息同步导致网络带宽压力和消耗很重，另外，没有扩展性可言，如果queue负载很重，加机器，新增的机器也包含了这个queue的所有数据，并没有办法线性扩展你的queue。此时，需要开启镜像集群模式，在rabbitmq管理控制台新增一个策略，将数据同步到指定数量的节点，然后你再次创建queue的时候，应用这个策略，就会自动将数据同步到其他的节点上去了。

Kafka 是 Apache 的子项目，是一个高性能跨语言的分布式发布/订阅消息队列系统（没有严格实现 JMS 规范的点对点模型，但可以实现其效果），在企业开发中有广泛的应用。高性能是其最大优势，劣势是消息的可靠性（丢失或重复），这个劣势是为了换取高性能，开发者可以以稍降低性能，来换取消息的可靠性。

一个Topic可以认为是一类消息，每个topic将被分成多个partition(区)，每个partition在存储层面是append log文件。任何发布到此partition的消息都会被直接追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），offset为一个long型数字，它是唯一标记一条消息。它唯一的标记一条消息。kafka并没有提供其他额外的索引机制来存储offset，因为在kafka中几乎不允许对消息进行“随机读写”。

Kafka和JMS（Java Message Service）实现(activeMQ)不同的是:即使消息被消费，消息仍然不会被立即删除。日志文件将会根据broker中的配置要求，保留一定的时间之后删除；比如log文件保留2天，那么两天后，文件会被清除，无论其中的消息是否被消费。kafka通过这种简单的手段，来释放磁盘空间，以及减少消息消费之后对文件内容改动的磁盘IO开支。

对于consumer而言，它需要保存消费消息的offset，对于offset的保存和使用，有consumer来控制；当consumer正常消费消息时，offset将会"线性"的向前驱动，即消息将依次顺序被消费。事实上consumer可以使用任意顺序消费消息，它只需要将offset重置为任意值。(offset将会保存在zookeeper中，参见下文)

kafka集群几乎不需要维护任何consumer和producer状态信息，这些信息有zookeeper保存；因此producer和consumer的客户端实现非常轻量级，它们可以随意离开，而不会对集群造成额外的影响。

partitions的设计目的有多个。最根本原因是kafka基于文件存储。通过分区，可以将日志内容分散到多个server上，来避免文件尺寸达到单机磁盘的上限，每个partiton都会被当前server(kafka实例)保存；可以将一个topic切分多任意多个partitions，来消息保存/消费的效率。此外越多的partitions意味着可以容纳更多的consumer，有效提升并发消费的能力。(具体原理参见下文)。

一个Topic的多个partitions，被分布在kafka集群中的多个server上；每个server(kafka实例)负责partitions中消息的读写操作；此外kafka还可以配置partitions需要备份的个数(replicas)，每个partition将会被备份到多台机器上，以提高可用性。

基于replicated方案，那么就意味着需要对多个备份进行调度；每个partition都有一个server为"leader"；leader负责所有的读写操作，如果leader失效，那么将会有其他follower来接管(成为新的leader)；follower只是单调的和leader跟进，同步消息即可。由此可见作为leader的server承载了全部的请求压力，因此从集群的整体考虑，有多少个partitions就意味着有多少个"leader"，kafka会将"leader"均衡的分散在每个实例上，来确保整体的性能稳定。

Producers

Producer将消息发布到指定的Topic中，同时Producer也能决定将此消息归属于哪个partition；比如基于"round-robin"方式或者通过其他的一些算法等。

Consumers

本质上kafka只支持Topic。每个consumer属于一个consumer group；反过来说，每个group中可以有多个consumer。发送到Topic的消息，只会被订阅此Topic的每个group中的一个consumer消费。

如果所有的consumer都具有相同的group，这种情况和queue模式很像；消息将会在consumers之间负载均衡。

如果所有的consumer都具有不同的group，那这就是"发布-订阅"；消息将会广播给所有的消费者。

在kafka中，一个partition中的消息只会被group中的一个consumer消费；每个group中consumer消息消费互相独立；我们可以认为一个group是一个"订阅"者，一个Topic中的每个partions，只会被一个"订阅者"中的一个consumer消费，不过一个consumer可以消费多个partitions中的消息。kafka只能保证一个partition中的消息被某个consumer消费时，消息是顺序的。事实上，从Topic角度来说，消息仍不是有序的。

Kafka的设计原理决定，对于一个topic，同一个group中不能有多于partitions个数的consumer同时消费，否则将意味着某些consumer将无法得到消息。

Guarantees

Kafka就比较适合高吞吐量并且允许少量数据丢失的场景，如果非要保证“消息可靠传输”，可以使用JMS。

Kafka Producer 消息发送有两种方式(配置参数 producertype)：

对于同步方式(producertype=sync)？Kafka Producer 消息发送有三种确认方式(配置参数 acks)：

kafka的设计初衷是希望作为一个统一的信息收集平台，能够实时的收集反馈信息，并需要能够支撑较大的数据量，且具备良好的容错能力。

持久性

kafka使用文件存储消息，这就直接决定kafka在性能上严重依赖文件系统的本身特性。且无论任何OS下，对文件系统本身的优化几乎没有可能。文件缓存/直接内存映射等是常用的手段。因为kafka是对日志文件进行append操作，因此磁盘检索的开支是较小的；同时为了减少磁盘写入的次数，broker会将消息暂时buffer起来，当消息的个数(或尺寸)达到一定阀值时，再flush到磁盘，这样减少了磁盘IO调用的次数。

性能

需要考虑的影响性能点很多，除磁盘IO之外，我们还需要考虑网络IO，这直接关系到kafka的吞吐量问题。kafka并没有提供太多高超的技巧；对于producer端，可以将消息buffer起来，当消息的条数达到一定阀值时，批量发送给broker；对于consumer端也是一样，批量fetch多条消息。不过消息量的大小可以通过配置文件来指定。对于kafka broker端，似乎有个sendfile系统调用可以潜在的提升网络IO的性能:将文件的数据映射到系统内存中，socket直接读取相应的内存区域即可，而无需进程再次copy和交换。其实对于producer/consumer/broker三者而言，CPU的开支应该都不大，因此启用消息压缩机制是一个良好的策略；压缩需要消耗少量的CPU资源，不过对于kafka而言，网络IO更应该需要考虑。可以将任何在网络上传输的消息都经过压缩。kafka支持gzip/snappy等多种压缩方式。

生产者

负载均衡: producer将会和Topic下所有partition leader保持socket连接；消息由producer直接通过socket发送到broker，中间不会经过任何“路由层“。事实上，消息被路由到哪个partition上，有producer客户端决定。比如可以采用“random““key-hash““轮询“等，如果一个topic中有多个partitions，那么在producer端实现“消息均衡分发“是必要的。

其中partition leader的位置(host:port)注册在zookeeper中，producer作为zookeeper client，已经注册了watch用来监听partition leader的变更事件。

异步发送：将多条消息暂且在客户端buffer起来，并将他们批量的发送到broker，小数据IO太多，会拖慢整体的网络延迟，批量延迟发送事实上提升了网络效率。不过这也有一定的隐患，比如说当producer失效时，那些尚未发送的消息将会丢失。

消费者

consumer端向broker发送“fetch”请求，并告知其获取消息的offset；此后consumer将会获得一定条数的消息；consumer端也可以重置offset来重新消费消息。

在JMS实现中，Topic模型基于push方式，即broker将消息推送给consumer端。不过在kafka中，采用了pull方式，即consumer在和broker建立连接之后，主动去pull(或者说fetch)消息；这中模式有些优点，首先consumer端可以根据自己的消费能力适时的去fetch消息并处理，且可以控制消息消费的进度(offset)；此外，消费者可以良好的控制消息消费的数量，batch fetch。

其他JMS实现，消息消费的位置是有prodiver保留，以便避免重复发送消息或者将没有消费成功的消息重发等，同时还要控制消息的状态。这就要求JMS broker需要太多额外的工作。在kafka中，partition中的消息只有一个consumer在消费，且不存在消息状态的控制，也没有复杂的消息确认机制，可见kafka broker端是相当轻量级的。当消息被consumer接收之后，consumer可以在本地保存最后消息的offset，并间歇性的向zookeeper注册offset。由此可见，consumer客户端也很轻量级。

对于JMS实现，消息传输担保非常直接:有且只有一次(exactly once)。

在kafka中稍有不同:

at most once: 消费者fetch消息，然后保存offset，然后处理消息；当client保存offset之后，但是在消息处理过程中出现了异常，导致部分消息未能继续处理。那么此后"未处理"的消息将不能被fetch到，这就是"at most once"。

at least once: 消费者fetch消息，然后处理消息，然后保存offset。如果消息处理成功之后，但是在保存offset阶段zookeeper异常导致保存操作未能执行成功，这就导致接下来再次fetch时可能获得上次已经处理过的消息，这就是"at least once"，原因offset没有及时的提交给zookeeper，zookeeper恢复正常还是之前offset状态。

exactly once: kafka中并没有严格的去实现(基于2阶段提交，事务)，我们认为这种策略在kafka中是没有必要的。

通常情况下“at-least-once”是我们首选。(相比at most once而言，重复接收数据总比丢失数据要好)。

kafka高可用由多个broker组成，每个broker是一个节点；

创建一个topic，这个topic会划分为多个partition，每个partition存在于不同的broker上，每个partition就放一部分数据。

kafka是一个分布式消息队列，就是说一个topic的数据，是分散放在不同的机器上，每个机器就放一部分数据。

在08版本以前，是没有HA机制的，就是任何一个broker宕机了，那个broker上的partition就废了，没法写也没法读，没有什么高可用性可言。

08版本以后，才提供了HA机制，也就是就是replica副本机制。每个partition的数据都会同步到其他的机器上，形成自己的多个replica副本。然后所有replica会选举一个leader出来，那么生产和消费都跟这个leader打交道，然后其他replica就是follower。

写的时候，leader会负责把数据同步到所有follower上去，读的时候就直接读leader上数据即可。

kafka会均匀的将一个partition的所有replica分布在不同的机器上，从而提高容错性。

如果某个broker宕机了也没事，它上面的partition在其他机器上都有副本的，如果这上面有某个partition的leader，那么此时会重新选举一个新的leader出来，大家继续读写那个新的leader即可。这就有所谓的高可用性了。

写数据的时候，生产者就写leader，然后leader将数据落地写本地磁盘，接着其他follower自己主动从leader来pull数据。一旦所有follower同步好数据了，就会发送ack给leader，leader收到所有follower的ack之后，就会返回写成功的消息给生产者。

消息丢失会出现在三个环节，分别是生产者、mq中间件、消费者：

RabbitMQ

Kafka

大体和RabbitMQ相同。

Rabbitmq

需要保证顺序的消息投递到同一个queue中，这个queue只能有一个consumer，如果需要提升性能，可以用内存队列做排队，然后分发给底层不同的worker来处理。

Kafka

写入一个partition中的数据一定是有序的。生产者在写的时候，可以指定一个key，比如指定订单id作为key，这个订单相关数据一定会被分发到一个partition中去。消费者从partition中取出数据的时候也一定是有序的，把每个数据放入对应的一个内存队列，一个partition中有几条相关数据就用几个内存队列，消费者开启多个线程，每个线程处理一个内存队列。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。

AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

那么，基于 RabbitMQ（消息队列）实现数据异步入库有什么好处呢？

RabbitMQ客户端模块

参考：

RabbitMQ基础概念详细介绍

消息队列

也称为 Warren (兔子窝) 模式。实现 rabbitMQ 的高可用集群，一般在并发和数据量不高的情况下，这种模式非常的好用且简单。

也就是一个主/备方案，主节点提供读写，备用节点不提供读写。如果主节点挂了，就切换到备用节点，原来的备用节点升级为主节点提供读写服务，当原来的主节点恢复运行后，原来的主节点就变成备用节点，和 activeMQ 利用 zookeeper 做主/备一样，也可以一主多备。

HaProxy 配置:

listen rabbitmq_cluster

bind 0000:567 # 配置 tcp 模式

mode tcp # 简单的轮询

balance roundrobin # 主节点 roundrobin 随机

server 你的76机器 hostname 1921681176:5672 check inter 5000 rise 2 fall 2

server 你的77机器 hostname 1921681177:5672 backup check inter 5000 rise 2 fall 2 # 备用节点

注意了，上面的 rabbitMQ 集群节点配置 # inter 每隔 5 秒对 mq 集群做健康检查， 2 次正确证明服务可用，2 次失败证明服务器不可用，并且配置主备机制

远程模式可以实现双活的一种模式，简称 shovel 模式，所谓的 shovel 就是把消息进行不同数据中心的复制工作，可以跨地域的让两个 MQ 集群互联，远距离通信和复制。

Shovel 就是我们可以把消息进行数据中心的复制工作，我们可以跨地域的让两个 MQ 集群互联。

如图所示，有两个异地的 MQ 集群（可以是更多的集群），当用户在地区 1 这里下单了，系统发消息到 1 区的 MQ 服务器，发现 MQ 服务已超过设定的阈值，负载过高，这条消息就会被转到地区 2 的 MQ 服务器上，由 2 区的去执行后面的业务逻辑，相当于分摊我们的服务压力。

在使用了 shovel 插件后，模型变成了近端同步确认，远端异步确认的方式，大大提高了订单确认速度，并且还能保证可靠性。

如上图所示，当我们的消息到达 exchange，它会判断当前的负载情况以及设定的阈值，如果负载不高就把消息放到我们正常的 warehouse_goleta 队列中，如果负载过高了，就会放到 backup_orders 队列中。backup_orders 队列通过 shovel 插件与另外的 MQ 集群进行同步数据，把消息发到第二个 MQ 集群上。

这是 rabbitMQ 比较早期的架构模型了，现在很少使用了。

shovel 集群的配置，首先启动 rabbitmq 插件，命令如下:

rabbitmq-plugins enable amqp_client

rabbitmq-plugins enable rabbitmq_shovel

在 /etc/rabbitmq/ 目录下创建 rabbitmqconfig 文件。注意，我们源服务器和目的地服务器都使用这个相同的配置文件。

具体配置如下

非常经典的 mirror 镜像模式，保证 100% 数据不丢失。在实际工作中也是用得最多的，并且实现非常的简单，一般互联网大厂都会构建这种镜像集群模式。

mirror 镜像队列，目的是为了保证 rabbitMQ 数据的高可靠性解决方案，主要就是实现数据的同步，一般来讲是 2 - 3 个节点实现数据同步。对于 100% 数据可靠性解决方案，一般是采用 3 个节点。

集群架构如下

如上图所示，用 KeepAlived 做了 HA-Proxy 的高可用，然后有 3 个节点的 MQ 服务，消息发送到主节点上，主节点通过 mirror 队列把数据同步到其他的 MQ 节点，这样来实现其高可靠。

也是实现异地数据复制的主流模式，因为 shovel 模式配置比较复杂，所以一般来说，实现异地集群的都是采用这种双活或者多活模型来实现的。这种模式需要依赖 rabbitMQ 的 federation 插件，可以实现持续的，可靠的 AMQP 数据通信，多活模式在实际配置与应用非常的简单。

rabbitMQ 部署架构采用双中心模式(多中心)，那么在两套(或多套)数据中心各部署一套 rabbitMQ 集群，各中心的rabbitMQ 服务除了需要为业务提供正常的消息服务外，中心之间还需要实现部分队列消息共享。

多活集群架构如下:

federation 插件是一个不需要构建 cluster ，而在 brokers 之间传输消息的高性能插件，federation 插件可以在 brokers 或者 cluster 之间传输消息，连接的双方可以使用不同的 users 和 virtual hosts，双方也可以使用不同版本的 rabbitMQ 和 erlang。federation 插件使用 AMQP 协议通信，可以接受不连续的传输。federation 不是建立在集群上的，而是建立在单个节点上的，如图上**的 rabbit node 3 可以与绿色的 node1、node2、node3 中的任意一个利用 federation 插件进行数据同步。

如上图所示，federation exchanges 可以看成 downstream 从 upstream 主动拉取消息，但是并不是拉取所有消息，必须是在 downstream 上已经明确定义 Bingdings 关系的 exchange，也就是有实际的物理 queue 来接收消息，才会从 upstream 拉取消息到 downstream 。

它使用 AMQP 协议实现代理间通信，downstream 会将绑定关系组合在一起，绑定/解绑命令将发送到 upstream 交换机。因此，federation exchange 只接收具有订阅的消息。

如果有一个消息生产者或者消息消费者通过amqp-client的客户端连接至节点1进行消息的发布或者订阅，那么此时的集群中的消息收发只与节点1相关。

如果消息生产者所连接的是节点2或者节点3，此时队列1的完整数据不在该两个节点上，那么在发送消息过程中这两个节点主要起了一个路由转发作用，根据这两个节点上的元数据转发至节点1上，最终发送的消息还是会存储至节点1的队列1上。

RabbitMQ 集群是一个或多个节点的逻辑分组，每个节点共享用户、虚拟主机、队列、交换器、绑定、运行时参数和其他分布式状态。

一些分布式系统有leader和follower节点。对于 RabbitMQ 来说， RabbitMQ集群中的所有节点都是平等的。

RabbitMQ 集群可以通过多种方式组成：

RabbitMQ 节点绑定到端口以接受客户端和 CLI 工具连接。其他进程和工具（例如 SELinux）可能会阻止 RabbitMQ 绑定到端口。发生这种情况时，节点将无法启动。

CLI工具、客户端库和 RabbitMQ 节点也会打开连接（客户端 TCP 套接字）。防火墙会阻止节点和 CLI 工具相互通信。确保可以访问以下端口：

RabbitMQ节点和 CLI 工具（例如 rabbitmqctl）使用 cookie 来确定它们是否被允许相互通信。为了让两个节点能够通信，它们必须具有相同的共享密钥，称为 Erlang cookie。cookie 只是一串最多 255 个字符的字母数字字符。它通常存储在本地文件中。该文件必须只能由所有者访问（例如具有 600 或类似的 UNIX 权限）。每个集群节点必须具有相同的 cookie。

在 UNIX 系统上，cookie通常是位于/var/lib/rabbitmq/erlangcookie（由服务器使用）和$HOME/erlangcookie（由 CLI 工具使用）。

RabbitMQ 节点使用主机名相互寻址

<!== 所有主机执行 ==>

默认配置文件/usr/lib/rabbitmq/lib/rabbitmq_server-3717/ebin/rabbitapp

<!== node01主机执行 ==>

<!== node02主机执行 ==>

<!== node03主机执行 ==>

<!== 所有主机执行 ==>

因RabbitMQ 集群元数据同步基于 cookie 共享方案实现

文件路径/var/lib/rabbitmq/erlangcookie

<!== node02、node03主机执行 ==>

<!== 任意主机执行 ==>

节点分为：磁盘节点及内存节点

RAM节点是一种特殊情况，可用于提高具有高队列、交换或绑定搅动的集群的性能。RAM节点不提供更高的消息速率。 官方建议在绝大多数情况下，仅使用磁盘节点。

如果一个集群中都是RAM节点，那么集群停止，将无法再次启动，并将丢失所有数据

官方提示：经典队列镜像将在未来版本中删除，考虑改用仲裁队列或非复制经典队列

每个镜像队列由一个领导副本和一个或多个镜像副本，leader被托管的节点成为leader节点。首先应用给定队列的所有操作在队列的leader节点上，然后传播到镜像。

如果承载队列的leader节点出现故障，则只要已同步，最旧的镜像将升级为新的leader。根据队列镜像参数，也可以升级未同步的镜像。

队列通过策略启用了镜像，策略模式的说明如下：

每当队列的策略发生变化时，它都保持其现有的镜像尽可能适用新策略。

设置的镜像队列可以通过开启的网页的管理端Admin->Policies，也可以通过命令。

管理端界面：

命令行：

为避免集群中的某个节点托管大多数leader队列，因此导致高负载，leader队列应合理均匀的分布在集群节点上。控制leader队列分布节点策略有三种，可以在rabbitmqconf文件中定义queue_master_locator参数设置

修改节点策略可能会导致现有的leader队列没有在新的策略中，为了防止消息丢失，RabbitMQ会保留现有的leader直到至少另一个镜像已同步，一旦发生同步，消费者将与leader断开连接，需要重新连接。

如果leader故障，其他镜像升级为leader过程如下：

如果消费者使用自动确认模式，那么消息可能会丢失。这与非镜像队列没有什么不同：代理认为消息一旦以自动确认模式发送给消费者，就会被确认。

如果客户端突然断开连接，则可能永远不会收到消息。在镜像队列的情况下，如果leader死亡，那些正在以自动确认模式发送给消费者的消息可能永远不会被这些客户端接收，并且不会被新leader重新排队。由于消费客户端连接到存活节点的可能性，消费者取消通知有助于识别此类事件何时发生。当然，在实践中，如果数据安全性不如吞吐量重要，那么自动确认模式是可行的方法。

节点可以随时加入集群。根据队列的配置，当节点加入集群时，队列可能会在新节点上添加镜像。此时，新镜像将是空的：它不会包含队列中任何现有的内容。这样的镜像将接收发布到队列的新消息，因此随着时间的推移将准确地表示镜像队列的尾部。随着消息从镜像队列中排出，新镜像丢失消息的队列头部的大小将缩小，直到最终镜像的内容与leader的内容完全匹配。在这一点上，镜像可以被认为是完全同步的。

新添加的镜像不提供添加镜像之前存在的队列内容的额外形式的冗余或可用性，除非队列已明确同步。由于在发生明确同步时队列变得无响应，因此最好允许正在从中排出消息的活动队列自然同步，并且仅明确同步非活动队列。

启用自动队列镜像时，请考虑所涉及队列的预期磁盘数据集 。具有庞大数据集（例如，数十 GB 或更多）的队列必须将其复制到新添加的镜像中，这会给集群资源（例如网络带宽和磁盘 I/O）带来很大的负载。

要查看镜像状态（是否同步）：

手动同步队列：

取消正在进行的同步：

如果你停止一个包含镜像队列leader的 RabbitMQ 节点，其他节点上的一些镜像将被提升为leader。如果继续停止节点，那么将到达一个镜像队列不再有镜像的点：它仅存在于一个节点上，且该节点上它是leader。如果它的最后一个剩余节点关闭，但是镜像队列被声明为持久的，那么队列中的持久消息将在该节点重新启动后继续存在。

然而，镜像目前无法知道它的队列内容是否已经偏离了它重新加入的leader。因此，当一个镜像重新加入一个镜像队列时， 它会丢弃已经拥有的任何持久本地内容并开始为空 。

默认情况下，RabbitMQ 将拒绝leader节点在受控关闭（即明确停止 RabbitMQ 服务或关闭操作系统）时提升非同步镜像，以避免消息丢失；相反，整个队列将关闭，就好像未同步的镜像不存在一样。

leader节点不受控制的关闭（即服务器或节点崩溃，或网络中断）仍将触发未同步镜像的提升。

如果您希望在所有情况下都让leader队列移动到未同步的镜像（即，您会选择队列的可用性而不是避免由于未同步的镜像升级而导致的消息丢失），那么将 ha-promote-on-shutdown 策略键设置为 always 而不是比它的默认值 when-synced 。

如果 ha-promote-on-failure 策略键设置为 when-synced ，则即使 ha-promote-on-shutdown 键设置为 always ，也不会提升未同步的镜像。这意味着如果leader节点发生故障，队列将变得不可用，直到leader恢复。如果leader队列永久丢失，队列将不可用，除非它被删除（这也将删除其所有内容）并重新声明。

当队列的所有镜像都关闭时，可能会丢失队列的leader。在正常操作中，队列关闭的最后一个节点将成为leader，该节点在再次启动时仍然是leader（因为它可能收到了其他镜像没有看到的消息）。

但是，当您调用 rabbitmqctl forget_cluster_node 时，RabbitMQ 将尝试为每个在我们忘记的节点上有其领导者的队列找到当前停止的镜像，并在它再次启动时“提升”该镜像成为新的领导者。如果有多个候选者，将选择最近停止的镜像。

重要的是要理解 RabbitMQ 只能在 forget_cluster_node 期间提升停止的镜像，因为任何再次启动的镜像都会清除它们的内容，如上面“停止节点和同步”中所述。因此在停止的集群中移除丢失的leader时，您必须在再次启动镜像之前调用 rabbitmqctl forget_cluster_node 。

我们知道常用的2款消息中间件是rabbitmq和kafka，他们2者都有什么各自的特点和应用场景呢？我们下面就聊一聊。

rabbitmq消息的发送，首先经过exchange，然后由exchange根据路由把消息投递到绑定的队列中，exchange有3中类型：

fanout：完全模式，消息会投递到和exchange绑定的所以队列中，使用于多个消费者应用的场景。

direct：直接模式，消息投递根据发送的routingkey和bindingkey完全匹配的队列中，使用单消费者场景。

topic：topic模式，消息投递模糊匹配路由规则，可以投递到多个匹配的队列中。

rabbitmq的消息可以持久化，也可以不持久化，消息消费完就被删除，不能重复消费。

rabbitmq cluster消息的存储，一个队列只能在一台机器上存储，无法实现分片存储。

kafka消息发送topic，topic可以多个分区，同一个topic的消息可以分片保存在不同的机器上，消息持久化存储在磁盘上，一个消息可以重复消费，不用像rabbitmq一样，有几个消费者来消费这个消息，就需要几个消息队列来存储这个消息。

rabbitmq是建立在AMQP上的企业消息系统。

以生产者消费者为模型而存在的一个消息队列

1、解耦

这是一个天然的解耦，实现了应用程序不再通过接口，你只需要调用消息队列的接口把结果存放在消息队列即可。