RabbitMQ

文章目录

Kafka、ActiveMQ、RabbitMQ、RocketMQ对比
ActiveMQRabbitMQRocketMQKafka单机吞吐量比 RabbitMQ低2.6w/s（消息做持久化）11.6w/s17.3w/s开发语言JavaErlangJavaScala/Java主要维护者ApacheMozilla/SpringAlibabaApache成熟度成熟成熟开源版本不够成熟比较成熟订阅形式点对点 (p2p)、广播（发布订阅）提供了4 种： direct、topic、headers、fanout。fanout就是广播模式基于topic/messageTag以及按照消息类型、属性进行正则匹配的发布订阅模式基于topic以及按照topic进行正则匹配的发布订阅模式持久化支持少量堆积支持少量堆积支持大量堆积支持大量堆积顺序消息不支持不支持支持支持性能稳定性好好一般较差集群方式支持简单集群模式，比如’主备’，对高级集群模式支持不好。支持简单集群，'复制’模式，对高级集群模式支持不好。常用多对’MasterSlave’模式，开源版本需手动切换Slave变成Master天然的‘LeaderSlave’无状态集群，每台服务器既是Master也是Slave管理界面一般较好一般无
RabbitMQ：延时低，微妙级延时，社区活跃度高，bug修复及时，而且提供了很友善的后台界面；用Erlang 语言开发，只熟悉 Java 的无法阅读源码和自行修复 bug。
RocketMQ：阿里维护的消息中间件，可以达到十万级的吞吐量，支持分布式事务。
Kafka：分布式的中间件，最大优点是其吞吐量高，一般运用于大数据系统的实时运算和日志采集的场景，功能简单，可靠性高，扩展性高；缺点是可能导致重复消费。

中小型公司，技术实力较为一般，技术挑战不是特别高，用RabbitMQ是不错的选择；大型公司，基础架构研发实力较强，用RocketMQ是很好的选择。如果是大数据领域的实时计算、日志采集等场景，用Kafka是业内标准的，社区活跃度很高。

一、初识RabbitMQ

RabbitMQ是一个由Erlang语言开发的AMQP的开源实现。

AMQP ：Advanced Message Queue，高级消息队列协议。它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。

1.1 RabbitMQ的特点

1、可靠性 RabbitMQ使用一些机制来保证消息的可靠性，如持久化、传输确认及发布确认等。
2、灵活的路由 在消息进入队列之前，通过交换器来路由消息。对于典型的路由功能，RabbitMQ己经提供了一些内置的交换器来实现。针对更复杂的路由功能，可以将多个交换器绑定在一起，也可以通过插件机制来实现自己的交换器。
3、扩展性 多个RabbitMQ节点可以组成一个集群，也可以根据实际业务情况动态地扩展集群中节点。
4、高可用性 队列可以在集群中的机器上设置镜像，使得在部分节点出现问题的情况下队列仍然可用。
5、支持多种协议 RabbitMQ除了原生支持AMQP协议，还支持STOMP、MQTT等多种消息中间件协议。
6、多语言客户端 RabbitMQ几乎支持所有常用语言，比如Java、Python、Ruby、PHP、C#、JavaScript等。
7、易用的管理界面 RabbitMQ提供了一个易用的用户界面，使得用户可以监控和管理消息、集群中的节点等。
8、插件机制 RabbitMQ提供了许多插件，以实现从多方面进行扩展，当然也可以编写自己的插件。

1.2 RabbitMQ基本概念*

Broker 表示消息队列服务器实体。
Exchange交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Queue 消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。
Binding 绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
Routing Key 路由关键字，exchange根据这个关键字进行消息投递。
VHost 可以理解为虚拟broker ，即mini-RabbitMQ server。其内部均含有独立的queue、exchange和binding等。重要的是，其拥有独立的权限系统，可以做到vhost范围的用户控制。
Publisher 消息的生产者，也是一个向交换器发布消息的客户端应用程序。
Channel 信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的 TCP 连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
Consumer 消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
Message 消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。
Connection 网络连接，比如一个 TCP 连接。

由Exchange、Queue、RoutingKey三个才能决定一个从Exchange到Queue的唯一的线路。

1.3 工作模式

RabbitMQ官网提供了七种队列模型，分别是：简单队列、工作队列、发布订阅、路由模式、主题模式、RPC模式、发布者确认模式。

1、simple模式（即最简单的收发模式） 1）消息产生消息，将消息放入队列。 2）消息的消费者监听消息队列，如果队列中有消息，就消费掉，消息被拿走后，自动从队列中删除(隐患：消息可能没有被消费者正确处理，已经从队列中消失了，造成消息的丢失，这里可以设置成手动的ack，但如果设置成手动ack，处理完后要及时发送ack消息给队列，否则会造成内存溢出)。
2、work工作模式(资源的竞争) 消息产生者将消息放入队列，消费者可以有多个，消费者1、消费者2同时监听同一个队列，消息被消费。C1、C2共同争抢当前的消息队列内容，谁先拿到谁负责消费消息(隐患：高并发情况下,默认会产生某一个消息被多个消费者共同使用，可以设置一个开关(syncronize)保证一条消息只能被一个消费者使用)。
3、publish/subscribe发布订阅(共享资源，类似广播) 1）每个消费者监听自己的队列； 2）生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。
4、routing路由模式 1）X代表交换机，生产者没有将消息直接发送到队列，而是先发送到了交换机。 2）路由模式的交换机类型为direct。 3）一个队列可以有多个消费者，但发送到队列的消息只能被其中一个消费。
5、topic 主题模式(路由模式的一种) 1）星号井号代表通配符。 2）星号代表多个单词,井号代表一个单。 3）路由功能添加模糊匹配。 4）消息产生者产生消息,把消息交给交换机。 5）交换机根据key的规则模糊匹配到对应的队列,由队列的监听消费者接收消息消费。

1.4 RabbitMQ详细概念*

RabbitMQ整体上是一个生产者与消费者模型，主要负责接收、存储和转发消息。
RabbitMQ的整体模型架构：

Producer：生产消息的一方。
Consumer：消费消息的一方。
消息一般由2部分组成：消息头（或者说是标签Label）和消息体。消息体也可以称为payLoad。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。生产者把消息交由RabbitMQ后，RabbitMQ会根据消息头把消息发送给感兴趣的Consumer。

1.4.1 Exchange(交换器)

在RabbitMQ中，消息并不是直接被投递到Queue(消息队列)中的，中间还必须经过Exchange这一层，Exchange会把消息分配到对应的Queue(消息队列) 中。
Exchange用来接收生产者发送的消息并将这些消息路由给服务器中的队列中，如果路由不到，或许会返回给 Producer(生产者) ，或许会被直接丢弃掉。这里可以将RabbitMQ中的交换器看作一个简单的实体。
RabbitMQ的Exchange有4种类型，不同的类型对应着不同的路由策略：direct(默认)、fanout、topic和headers，不同类型的Exchange转发消息的策略有所区别。

1.4.2 Binding(绑定)

生产者将消息发给交换器的时候，一般会指定一个RoutingKey(路由键)，用来指定这个消息的路由规则，而这个RoutingKey需要与交换器类型和绑定键(BindingKey)联合使用才能最终生效。
RabbitMQ中通过Binding(绑定)将Exchange与Queue(消息队列)关联起来，在绑定的时候一般会指定一个BindingKey(绑定建)。这样RabbitMQ就知道如何正确将消息路由到队列了。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。Exchange和Queue的绑定可以是多对多的关系。
Binding(绑定) 示意图：

生产者将消息发送给交换器时，需要一个RoutingKey，当BindingKey和RoutingKey相匹配时，消息会被路由到对应的队列中。在绑定多个队列到同一个交换器的时候，这些绑定允许使用相同的BindingKey。BindingKey并不是在所有的情况下都生效，它依赖于交换器类型，比如fanout类型的交换器就会无视，而是将消息路由到所有绑定到该交换器的队列中。

1.4.3 Queue(消息队列)

Queue用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。
RabbitMQ中消息只能存储在队列中，这一点和Kafka这种消息中间件相反。Kafka将消息存储在topic（主题）这个逻辑层面，而相对应的队列逻辑只是topic实际存储文件中的位移标识。

RabbitMQ的生产者生产消息并最终投递到队列中，消费者可以从队列中获取消息并消费

。
多个消费者可以订阅同一个队列，这时队列中的消息会被平均分摊（Round-Robin，即轮询）给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，这样避免的消息被重复消费。
RabbitMQ不支持队列层面的广播消费，如果有广播消费的需求，需要在其上进行二次开发，这样会很麻烦，不建议这样做。

1.4.4 Broker（消息中间件的服务节点）

对于RabbitMQ来说，一个RabbitMQ Broker可以简单地看作一个RabbitMQ服务节点，或者RabbitMQ服务实例。大多数情况下也可以将一个RabbitMQ Broker看作一台RabbitMQ服务器。
生产者将消息存入RabbitMQ Broker，以及消费者从Broker中消费数据的整个流程。

1.4.5 Exchange Types(交换器类型)

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种。

fanout fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中，不需要做任何判断操作，所以fanout类型是所有的交换机类型里面速度最快的。 fanout类型常用来广播消息。
direct direct类型的Exchange路由规则也很简单，它会把消息路由到那些Bindingkey与RoutingKey完全匹配的Queue中。以上图为例，如果发送消息的时候设置路由键为“warning”，那么消息会路由到Queue1和Queue2。如果在发送消息的时候设置路由键为"Info”或者"debug”，消息只会路由到Queue2。如果以其他的路由键发送消息，则消息不会路由到这两个队列中。 direct类型常用在处理有优先级的任务，根据任务的优先级把消息发送到对应的队列，这样可以指派更多的资源去处理高优先级的队列。
topic 前面讲到direct类型的交换器路由规则是完全匹配BindingKey和RoutingKey ，但是这种严格的匹配方式在很多情况下不能满足实际业务的需求。topic类型的交换器在匹配规则上进行了扩展，它与 direct类型的交换器相似，也是将消息路由到 BindingKey 和 RoutingKey 相匹配的队列中，但这里的匹配规则有些不同，它约定：

RoutingKey 为一个点号“．”分隔的字符串（被点号“．”分隔开的每一段独立的字符串称为一个单词），如 “com.rabbitmq.client”、“java.util.concurrent”、“com.hidden.client”；
BindingKey 和 RoutingKey 一样也是点号“．”分隔的字符串；
BindingKey 中可以存在两种特殊字符串“”和“#”，用于做模糊匹配，其中“”用于匹配一个单词，“#”用于匹配多个单词(可以是零个)。

以上图为例：

路由键为 “com.rabbitmq.client” 的消息会同时路由到 Queuel 和 Queue2;
路由键为 “com.hidden.client” 的消息只会路由到 Queue2 中；
路由键为 “com.hidden.demo” 的消息只会路由到 Queue2 中；
路由键为 “java.rabbitmq.demo” 的消息只会路由到Queuel中；
路由键为 “java.util.concurrent” 的消息将会被丢弃或者返回给生产者（需要设置 mandatory 参数），因为它没有匹配任何路由键。

headers headers类型的交换器不依赖于路由键的匹配规则来路由消息，而是根据发送的消息内容中的 headers属性进行匹配。在绑定队列和交换器时制定一组键值对，当发送消息到交换器时，RabbitMQ会获取到该消息的 headers（也是一个键值对的形式)'对比其中的键值对是否完全匹配队列和交换器绑定时指定的键值对，如果完全匹配则消息会路由到该队列，否则不会路由到该队列。headers 类型的交换器性能会很差，而且也不实用，基本上不会看到它的存在。

二、相关问题

2.1 RabbitMQ上的一个queue中存放的message是否有数量限制

可以认为是无限制，因为限制取决于机器器的内存，但是消息过多会导致处理效率的下降。

2.2 如何确保消息正确地发送至RabbitMQ*

RabbitMQ使用发送方确认模式，确保消息正确地发送到RabbitMQ。

发送方确认模式：将信道设置成confirm模式（发送方确认模式），则所有在信道上发布的消息都会被指派一个唯一的ID。一旦消息被投递到目的队列后，或者消息被写入磁盘后（可持久化的消息），信道会发送一个确认给生产者（包含消息唯一ID）。如果RabbitMQ发生内部错误从而导致消息丢失，会发送一条nack（not acknowledged，未确认）消息。

发送确认模式是异步的，生产者应用程序在等待确认的同时，可以继续发送消息。当确认消息到达生产者应用程序，生产者应用程序的回调方法就会被触发来处理确认消息。

2.3 如何确保消息接收方消费了消息

接收方消息确认机制：消费者接收每一条消息后都必须进行确认（消息接收和消息确认是两个不同操作）。只有消费者确认了消息，RabbitMQ才能安全地把消息从队列中删除。
这里并没有用到超时机制，RabbitMQ仅通过Consumer的连接中断来确认是否需要重新发送消息。也就是说，只要连接不中断，RabbitMQ给了Consumer足够长的时间来处理理消息。

特殊情况：

如果消费者接收到消息，在确认之前断开了连接或取消订阅，RabbitMQ会认为消息没有被分发，然后重新分发给下个订阅的消费者。（可能存在消息重复消费的隐患，需要根据bizId去重）

如果消费者接收到消息却没有确认消息，连接也未断开，则RabbitMQ认为该消费者繁忙，将不会给该消费者分发更多的消息。

2.4 如何避免消息重复投递或重复消费*

在消息生产时，MQ内部针对每条生产者发送的消息生成一个inner-msg-id，作为去重和幂等的依据（消息投递失败并重传），避免重复的消息进入队列；
在消息消费时，要求消息体中必须要有一个bizId（对于同一业务全局唯一，如支付ID、订单ID、帖子ID等）作为去重和幂等的依据，避免同一条消息被重复消费。

2.5 消息基于什么传输

由于TCP连接的创建和销毁开销较大，且并发数受系统资源限制，会造成性能瓶颈。RabbitMQ使用信道的方式来传输数据。信道是建立在真实的TCP连接内的虚拟连接，且每条TCP连接上的信道数量量没有限制。

RabbitMQ采用类似NIO（Non-blocking I/O）做法，选择TCP连接复用，不仅可以减少性能开销，同时也便于管理。

每个线程把持一个信道，所以信道复用了Connection的TCP连接。同时RabbitMQ可以确保每个线程的私密性，就像拥有独立的连接一样。

2.6 消息怎么路由

从概念上来说，消息路路由必须有三部分：交换器、路由、绑定。生产者把消息发布到交换器器上；绑定决定了消息如何从交换器路由到特定的队列；消息最终到达队列，并被消费者接收。

1、消息发布到交换器时，消息将拥有一个路由键（routing key），在消息创建时设定。
2、通过队列路由键，可以把队列绑定到交换器器上。
3、消息到达交换器后，RabbitMQ会将消息的路路由键与队列列的路路由键进行匹配（针对不同的交换器有不同的路由规则）。
4、如果能够匹配到队列列，则消息会投递到相应队列列中；如果不能匹配到任何队列列，消息将进入 “黑洞”。

消息提供方 >>> 路由 >>> 一至多个队列。
消息发布到交换器时，消息将拥有一个路由键（routing key），在消息创建时设定。
通过队列路由键，可以把队列绑定到交换器上。
消息到达交换器后，RabbitMQ 会将消息的路由键与队列的路由键进行匹配（针对不同的交换器有不同的路由规则）；
常用的交换器主要分为一下三种：

fanout：如果交换器收到消息，将会广播到所有绑定的队列上。
direct：如果路由键完全匹配，消息就被投递到相应的队列。
topic：可以使来自不同源头的消息能够到达同一个队列。使用topic交换器时，可以使用通配符。

2.7 如何确保消息不丢失*

1、生产者弄丢了数据 可以选择用 RabbitMQ 提供的事务功能，就是生产者发送数据之前开启 RabbitMQ 事务channel.txSelect ，然后发送消息，如果消息没有成功被 RabbitMQ 接收到，那么生产者会收到异常报错，此时就可以回滚事务 channel.txRollback ，然后重试发送消息；如果收到了消息，那么可以提交事务 channel.txCommit 。伪代码：

// 开启事务
channel txSelect
try{// 发送消息}catch(Exception){
   channel txRollback
   // 再次重发这条消息}// 提交事务
channel txCommit

问题是，RabbitMQ 事务机制（同步）一搞，基本上吞吐量会下来，因为太耗性能。
一般来说，如果你要确保说写 RabbitMQ 的消息别丢，可以开启 confirm 模式，在生产者那里设置开启 confirm 模式之后，你每次写的消息都会分配一个唯一的 id，然后如果写入了 RabbitMQ 中，RabbitMQ 会给你回传一个 ack 消息，告诉你说这个消息 ok 了。如果RabbitMQ 没能处理这个消息，会回调你的一个 nack 接口，告诉你这个消息接收失败，你可以重试。而且你可以结合这个机制自己在内存里维护每个消息 id 的状态，如果超过一定时间还没接收到这个消息的回调，那么你可以重发。
事务机制和 confirm 机制最大的不同在于，事务机制是同步的，你提交一个事务之后会阻塞在那儿，但是 confirm 机制是异步的，你发送个消息之后就可以发送下一个消息，然后那个消息 RabbitMQ 接收了之后会异步回调你的一个接口通知你这个消息接收到了。所以一般在生产者这块避免数据丢失，都是用 confirm 机制的。

2、RabbitMQ弄丢了数据 这个你必须开启 RabbitMQ 的持久化，就是消息写入之后会持久化到磁盘，哪怕是 RabbitMQ 自己挂了，恢复之后会自动读取之前存储的数据，一般数据不会丢。除非极其罕见的是，RabbitMQ 还没持久化，自己就挂了，可能导致少量数据丢失，但是这个概率较小。设置持久化有两个步骤：

创建 queue 的时候将其设置为持久化。这样就可以保证 RabbitMQ 持久化 queue 的元数据，但是它是不会持久化 queue 里的数据的。

第二个是发送消息的时候将消息的 deliveryMode 设置为 2。就是将消息设置为持久化的，此时 RabbitMQ 就会将消息持久化到磁盘上去。

必须要同时设置这两个持久化才行，RabbitMQ 哪怕是挂了，再次重启，也会从磁盘上重启恢复queue，恢复这个 queue 里的数据。
注意，哪怕是你给 RabbitMQ 开启了持久化机制，也有一种可能，就是这个消息写到了RabbitMQ 中，但是还没来得及持久化到磁盘上，结果不巧，此时 RabbitMQ 挂了，就会导致内存里的一点点数据丢失。所以，持久化可以跟生产者那边的 confirm 机制配合起来，只有消息被持久化到磁盘之后，才会通知生产者 ack 了，所以哪怕是在持久化到磁盘之前，RabbitMQ 挂了，数据丢了，生产者收不到 ack ，你也是可以自己重发的。

3、消费端弄丢了数据 RabbitMQ 如果丢失了数据，主要是因为你消费的时候，刚消费到，还没处理，结果进程挂了，比如重启了，那么就尴尬了，RabbitMQ 认为你都消费了，这数据就丢了。这个时候得用 RabbitMQ 提供的 ack 机制，简单来说，就是你必须关闭 RabbitMQ 的自动ack ，可以通过一个 api 来调用就行，然后每次你自己代码里确保处理完的时候，再在程序里 ack 一把。这样的话，如果你还没处理完，不就没有 ack 了？那 RabbitMQ 就认为你还没处理完，这个时候 RabbitMQ 会把这个消费分配给别的 consumer 去处理，消息是不会丢的。

2.8 如何保证消息的顺序性*

顺序会错乱的场景：RabbitMQ：一个queue，多个consumer。比如，生产者向 RabbitMQ 里发送了三条数据，顺序依次是 data1/data2/data3，压入的是 RabbitMQ 的一个内存队列。有三个消费者分别从MQ 中消费这三条数据中的一条，结果消费者2先执行完操作，把 data2 存入数据库，然后是 data1/data3。这时顺序就出了问题。

解决：拆分多个 queue，每个 queue 一个 consumer，就是多一些 queue 而已，确实是麻烦点；或者就一个 queue 但是对应一个 consumer，然后这个 consumer 内部用内存队列做排队，然后分发给底层不同的 worker 来处理。

2.9 RabbitMQ交换器有哪些类型*

fanout交换器：它会把所有发送到该交换器的消息路由到所有与该交换器绑定的队列列中；
direct交换器：direct类型的交换器路由规则很简单，它会把消息路由到哪些BindingKey和RoutingKey完全匹配的队列中；
topic交换器：匹配规则比direct更灵活。
headers交换器：根据发送消息内容的headers属性进行匹配（由于性能很差，不实用）。

常用的交换器：

1、direct：如果路由键完全匹配，消息就被投递到相应的队列。
2、fanout：如果交换器收到消息，将会播到所有绑定的队列上。
3、topic：可以使来自不同源头的消息能够到达同一个队列。使用topic交换器时，可以使用通配符，比如：“*” 匹配特定位置的任意文本， “.” 把路由键分为了几个部分，“#” 匹配所有规则等。特别注意：发往topic交换器的消息不能随意的设置选择键（routing_key），必须是由"."隔开的一系列的标识符组成。

2.10 RabbitMQ如何保证数据一致性

生产者确认机制：消息持久化后异步回调通知生产者，保证消息已经发出去；
消息持久化：设置消息持久化；
消费者确认机制：消费者成功消费消息之后，手动确认，保证消息已经消费。

2.11 RabbitMQ结构*

Broker：简单来说就是消息队列列服务器器实体。
Exchange：消息交换机，它指定消息按什什么规则，路路由到哪个队列。
Queue：消息队列列载体，每个消息都会被投入到一个或多个队列列。
Binding：绑定，它的作用就是把exchange和queue按照路路由规则绑定起来。
Routing Key：路路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

2.12 RabbitMQ队列与消费者的关系

一个队列可以绑定多个消费者；
消息默认以循环的方式发送给消费者；
消费者收到消息默认自动确认，也可以改成手动确认。

2.13 如何保证消息的可靠传输

数据的丢失问题，可能出现在生产者、MQ、消费者中。

生产者丢失 生产者将数据发送到RabbitMQ的时候，可能数据就在半路给搞丢了，因为网络问题啥的，都有可能。此时可以选择用RabbitMQ提供的事务功能，就是生产者发送数据之前开启RabbitMQ事务，然后发送消息。如果消息没有成功被RabbitMQ接收到，那么生产者会收到异常报错，此时就可以回滚事，然后重试发送消息；如果收到了消息，那么可以提交事务。这样做吞吐量会下来，因为太耗性能。所以你要确保说写RabbitMQ的消息别丢，可以开启confirm模式，在生产者那里设置开启confirm模式之后，你每次写的消息都会分配一个唯一的id，然后如果写入了RabbitMQ中，RabbitMQ会给你回传一个ack消息，告诉你说这个消息ok 了。如果RabbitMQ没能处理这个消息，会回调你一个nack接口，告诉你这个消息接收失败，你可以重试。而且你可以结合这个机制自己在内存里维护每个消息id的状态，如果超过一定时间还没接收到这个消息的回调，那么你可以重发。事务机制和cnofirm机制最大的不同在于，事务机制是同步的，你提交一个事务之后会阻塞在那儿，但是confirm机制是异步的，你发送个消息之后就可以发送下一个消息，然后那个消息RabbitMQ接收了之后会异步回调你一个接口通知你这个消息接收到了。所以一般在生产者这块避免数据丢失，都是用confirm机制的。
MQ中丢失 RabbitMQ自己弄丢了数据，这个你必须开启RabbitMQ的持久化，就是消息写入之后会持久化到磁盘，哪怕是RabbitMQ挂了，恢复之后会自动读取之前存储的数据，一般数据不会丢。设置持久化有两个步骤：创建queue的时候将其设置为持久化，这样就可以保证RabbitMQ持久化queue的元数据，但是不会持久化queue里的数据。第二个是发送消息的时候将消息的deliveryMode设置为2，就是将消息设置为持久化的，此时RabbitMQ就会将消息持久化到磁盘上去。必须要同时设置这两个持久化才行，RabbitMQ哪怕是挂了，再次重启，也会从磁盘上重启恢复queue，恢复这个queue里的数据。持久化可以跟生产者那边的confirm机制配合起来，只有消息被持久化到磁盘之后，才会通知生产者ack了，所以哪怕是在持久化到磁盘之前，RabbitMQ挂了，数据丢了，生产者收不到ack，你也是可以自己重发的。注意，哪怕是你给 RabbitMQ开启了持久化机制，也有一种可能，就是这个消息写到了RabbitMQ中，但是还没来得及持久化到磁盘上，结果不巧，此时RabbitMQ 挂了，就会导致内存里的一点点数据丢失。
消费端丢失 消费的时候，还没处理，结果进程挂了，比如重启了，RabbitMQ认为你都消费了，这数据就丢了。这个时候得用RabbitMQ提供的ack机制。简单来说，就是你关闭RabbitMQ的自动ack，可以通过一个api来调用就行，然后每次你自己代码里确保处理完的时候，再在程序里ack一把。这样的话，如果你还没处理完，不就没有ack？那RabbitMQ就认为你还没处理完，这个时候RabbitMQ会把这个消费分配给别的consumer去处理，消息是不会丢的。

2.14 如何解决消息队列的延时以及过期失效问题

消息积压处理办法：临时紧急扩容 先修复consumer的问题，确保其恢复消费速度，然后将现有cnosumer都停掉。新建一个topic，partition是原来的 10 倍，临时建立好原先10倍的queue数量。然后写一个临时的分发数据的consumer程序，这个程序部署上去消费积压的数据，消费之后不做耗时的处理，直接均匀轮询写入临时建立好的10倍数量的queue。接着临时征用10倍的机器来部署consumer，每一批consumer消费一个临时queue的数据。这种做法相当于是临时将queue资源和consumer资源扩大10倍，以正常的10倍速度来消费数据。等快速消费完积压数据之后，得恢复原先部署的架构，重新用原先的consumer 机器来消费消息。
MQ中消息失效 假设你用的是RabbitMQ，RabbtiMQ是可以设置过期时间的，也就是TTL。如果消息在queue中积压超过一定的时间就会被RabbitMQ给清理掉，这个数据就没了。那这就是第二个坑了。这就不是说数据会大量积压在mq里，而是大量的数据会直接搞丢。我们可以采取一个方案，就是批量重导，数据大量积压的时候，我们当时就直接丢弃数据了，然后等过了高峰期以后，就开始写程序，将丢失的那批数据，写个临时程序，一点一点的查出来，然后重新灌入mq里面去，把丢的数据给他补回来。比如1 万个订单积压在mq里面，没有处理，其中1000个订单都丢了，你只能手动写程序把那1000个订单给查出来，手动发到mq里去再补一次。

2.15 怎么保证消息队列的高可用*

RabbitMQ是基于主从（非分布式）做高可用性的。
RabbitMQ有三种模式：单机模式、普通集群模式、镜像集群模式。

单机模式 Demo级别的。
普通集群模式（无高可用性） 普通集群模式，意思就是在多台机器上启动多个RabbitMQ实例，每个机器启动一个。你创建的queue，只会放在一个RabbitMQ实例上，但是每个实例都同步queue的元数据（元数据可以认为是queue的一些配置信息，通过元数据，可以找到queue所在实例）。你消费的时候，实际上如果连接到了另外一个实例，那么那个实例会从queue所在实例上拉取数据过来。这种方式确实很麻烦，也没做到所谓的分布式，就是个普通集群。因为这导致你要么消费者每次随机连接一个实例然后拉取数据，要么固定连接那个queue所在实例消费数据，前者有数据拉取的开销，后者导致单实例性能瓶颈。而且如果那个放queue的实例宕机了，会导致接下来其他实例就无法从那个实例拉取，如果你开启了消息持久化，让RabbitMQ落地存储消息的话，消息不一定会丢，得等这个实例恢复了，然后才可以继续从这个queue拉取数据。这方案没有什么所谓的高可用性，主要是提高吞吐量，就是说让集群中多个节点来服务某个queue的读写操作。
镜像集群模式（高可用性） 在镜像集群模式下，你创建的queue，无论元数据还是queue里的消息都会存在于多个实例上，就是说，每个RabbitMQ节点都有这个queue的一个完整镜像，包含queue的全部数据的意思。然后每次你写消息到queue的时候，都会自动把消息同步到多个实例的queue上。如何开启这个镜像集群模式呢？其实很简单，RabbitMQ有很好的管理控制台，就是在后台新增一个策略，这个策略是镜像集群模式的策略，指定的时候是可以要求数据同步到所有节点的，也可以要求同步到指定数量的节点，再次创建queue的时候，应用这个策略，就会自动将数据同步到其他的节点上去了。

标签： rabbitmq 分布式

本文转载自: https://blog.csdn.net/m0_37741420/article/details/134904050
版权归原作者 解梦者 所有，如有侵权，请联系我们删除。