跳到主要内容

理论

RabbitMQ交换器有哪些类型？​

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

常⽤的交换器主要分为以下三种：

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

如何确保消息不丢失？​

消息持久化的前提是：将交换器/队列的durable属性设置为true，表示交换器/队列是持久交换器/队列，在服务器崩溃或重启之后不需要重新创建交换器/队列（交换器/队列会⾃动创建）。

如果消息想要从Rabbit崩溃中恢复，那么消息必须：

  1. 在消息发布前，通过把它的 “投递模式” 选项设置为2（持久）来把消息标记成持久化
  2. 将消息发送到持久交换器
  3. 消息到达持久队列

RabbitMQ确保持久性消息能从服务器重启中恢复的⽅式是，将它们写⼊磁盘上的⼀个持久化⽇志⽂件，当发布⼀条持久性消息到持久交换器上时，Rabbit会在消息提交到⽇志⽂件后才发送响应（如果消息路由到了⾮持久队列，它会⾃动从持久化⽇志中移除）。⼀旦消费者从持久队列中消费了⼀条持久化消息，RabbitMQ会在持久化⽇志中把这条消息标记为等待垃圾收集。如果持久化消息在被消费之前RabbitMQ重启，那么Rabbit会⾃动重建交换器和队列（以及绑定），并重播持久化⽇志⽂件中的消息到合适的队列或者交换器上。

消息如何分发？​

若该队列⾄少有⼀个消费者订阅，消息将以循环（round-robin）的方式发送给消费者。每条消息只会分发给⼀个订阅的消费者（前提是消费者能够正常处理消息并进行确认）。

消息怎么路由？​

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

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

如何避免消息重复投递或重复消费？​

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

如何确保消息正确地发送⾄RabbitMQ？​

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

**发送⽅确认模式：**将信道设置成confirm模式（发送⽅确认模式），则所有在信道上发布的消息

都会被指派⼀个唯⼀的ID。⼀旦消息被投递到⽬的队列后，或者消息被写⼊磁盘后（可持久化

的消息），信道会发送⼀个确认给⽣产者（包含消息唯⼀ID）。如果RabbitMQ发⽣内部错误

从⽽导致消息丢失，会发送⼀条nack（not acknowledged，未确认）消息。

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

如何确保消息接收⽅消费了消息？​

**接收⽅消息确认机制：**消费者接收每一条消息后都必须进行确认（消息接收和消息确认是两个不同操作）。只有消费者确认了消息，RabbitMQ才能安全地把消息从队列中删除。

这⾥并没有⽤到超时机制，RabbitMQ仅通过Consumer的连接中断来确认是否需要重新发送消息。也就是说，只要连接不中断，RabbitMQ给了Consumer⾜够⻓的时间来处理消息。

特殊情况：

  1. 如果消费者接收到消息，在确认之前断开了连接或取消订阅，RabbitMQ会认为消息没有被分发，然后重新分发给下一个订阅的消费者。（可能存在消息重复消费的隐患，需要根据bizId去重）
  2. 如果消费者接收到消息却没有确认消息，连接也未断开，则RabbitMQ认为该消费者繁忙，将不会给该消费者分发更多的消息。

上千万条消息在mq中积压了几个小时还没解决：​

  1. 先修复consumer的问题，确保其恢复消费速度，然后将现有consumer都停掉；
  2. 新建⼀个topic，partition是原来的10倍，临时建⽴好原先10倍或者20倍的queue数量；
  3. 然后写⼀个临时的分发数据的consumer程序，这个程序部署上去消费积压的数据；消费之后不做耗时的处理，直接均匀轮询写⼊临时建⽴好的10倍数量的queue；
  4. 接着临时征⽤10倍的机器来部署consumer，每⼀批consumer消费⼀个临时queue的数据；
  5. 这种做法相当于是临时将queue资源和consumer资源扩⼤10倍，以正常的10倍速度来消费数据；
  6. 等快速消费完积压数据之后，得恢复原先部署架构，重新⽤原先的consumer机器来消费消息。

总结：

  1. 修复并停掉consumer；
  2. 新建⼀个topic，partition是原来的10倍，建⽴临时queue，数量是原来的10倍或20倍；
  3. 写临时consumer程序，临时征⽤10倍的机器去消费数据；
  4. 消费完成之后，恢复原先consumer；

rabbitmq设置过期时间，部分消息丢失：​

采取批量重导⽅法：将丢失的那批数据查询导⼊到mq⾥⾯。