RabbitMQ - 介绍

如题所述

第1个回答  2022-08-01

各组件解释如下:

AMQP 消息的路由中增加了 Exchange 和 Binding 的角色。生产者把消息发布到 Exchange 上,消息最终到达队列并被消费者接收,而 Binding 决定交换器的消息应该发送到那个队列。

Exchange分发消息时根据类型的不同分发策略有区别,目前共四种类型:direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键,此外 headers 交换器和 direct 交换器完全一致,但性能差很多,目前几乎用不到了,所以直接看另外三种类型:

消息中的路由键(routing key)如果和 Binding 中的 binding key 一致, 交换器就将消息发到对应的队列中。路由键与队列名完全匹配,如果一个队列绑定到交换机要求路由键为"dog",则只转发 routing key 标记为"dog"的消息,不会转发"dog.puppy",也不会转发"dog.guard"等等。它是完全匹配、单播的模式。

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键,只是简单的将队列绑定到交换器上,每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播,每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。

topic 交换器通过模式匹配分配消息的路由键属性,将路由键和某个模式进行匹配,此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词,这些单词之间用点隔开。它同样也会识别两个通配符:符号"#"和符号" "。
"#"匹配0个或多个单词,"
"匹配不多不少一个单词。

RabbbitMQ 的分发机制非常适合扩展,而且它是专门为并发程序设计的,如果现在 load 加重,那么只需要创建更多的 Consumer 来进行任务处理。

在实际应用中,可能会发生消费者收到 Queue 中的消息,但没有处理完成就宕机(或出现其他意外)的情况,这种情况下就可能会导致消息丢失。为了避免这种情况发生,我们可以要求消费者在消费完消息后发送一个回执给 RabbitMQ,RabbitMQ 收到消息回执(Message acknowledgment)后才将该消息从 Queue 中移除;如果 RabbitMQ 没有收到回执并检测到消费者的 RabbitMQ 连接断开,则 RabbitMQ 会将该消息发送给其他消费者(如果存在多个消费者)进行处理。这里不存在 timeout 概念,一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者,除非它的 RabbitMQ 连接断开。 这里会产生另外一个问题,如果我们的开发人员在处理完业务逻辑后,忘记发送回执给 RabbitMQ,这将会导致严重的 bug——Queue 中堆积的消息会越来越多;消费者重启后会重复消费这些消息并重复执行业务逻辑。
另外 pub message 是没有 ack 的。

如果我们希望即使在 RabbitMQ 服务重启的情况下,也不会丢失消息,我们可以将 Queue 与 Message 都设置为可持久化的(durable),这样可以保证绝大部分情况下我们的 RabbitMQ 消息不会丢失。但依然解决不了小概率丢失事件的发生(比如 RabbitMQ 服务器已经接收到生产者的消息,但还没来得及持久化该消息时 RabbitMQ 服务器就断电了),如果我们需要对这种小概率事件也要管理起来,那么我们要用到事务。由于这里仅为 RabbitMQ 的简单介绍,所以这里将不讲解 RabbitMQ 相关的事务。
要持久化队列 queue 的持久化需要在声明时指定 durable=True;
这里要注意,队列的名字一定要是 Broker 中不存在的,不然不能改变此队列的任何属性.
队列和交换机有一个创建时候指定的标志 durable,durable 的唯一含义就是具有这个标志的队列和交换机会在重启之后重新建立,它不表示说在队列中的消息会在重启后恢复。

消息持久化包括 3 部分

如果 exchange 和 queue 都是持久化的,那么它们之间的 binding 也是持久化的,如果 exchange 和 queue 两者之间有一个持久化,一个非持久化,则不允许建立绑定.
注意:一旦创建了队列和交换机,就不能修改其标志了,例如,创建了一个 non-durable 的队列,然后想把它改变成 durable 的,唯一的办法就是删除这个队列然后重现创建。

你可能也注意到了,分发机制不是那么优雅,默认状态下,RabbitMQ 将第 n 个 Message 分发给第 n 个 Consumer。n 是取余后的,它不管 Consumer 是否还有 unacked Message,只是按照这个默认的机制进行分发.
那么如果有个 Consumer 工作比较重,那么就会导致有的 Consumer 基本没事可做,有的 Consumer 却毫无休息的机会,那么,Rabbit 是如何处理这种问题呢?

RabbitMQ 使用 ProtoBuf 序列化消息,它可作为 RabbitMQ 的 Message 的数据格式进行传输,由于是结构化的数据,这样就极大的方便了 Consumer 的数据高效处理,当然也可以使用 XML,与 XML 相比,ProtoBuf 有以下优势:

相似回答