ToB企服应用市场:ToB评测及商务社交产业平台

标题: RabbitMQ 基础概念与架构设计及工作机制学习总结 [打印本页]

作者: 美食家大橙子 时间: 2024-8-20 21:49
标题: RabbitMQ 基础概念与架构设计及工作机制学习总结
什么是RabbitMQ

MQ全称为Message Queue，即消息队列. 它也是一个队列，依照FIFO原则。RabbitMQ则是一个开源的消息中间件，由erlang语言开发，基于AMQP协议实现的一个软件产品，提供应用程序之间的通信方法，在分布式体系开发中广泛应用。
AMQP协议

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，一个提供统一消息服务的应用层标准高级消息队列协议，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件差别产品，差别的开发语言等条件的限定。
RabbitMQ基本架构

消息队列服务, 通常会涉及三个概念：消息生产者(简称生产者)，消息队列，消息消费者(简称消费者)。
RabbitMQ 在这个基础上, 多做了一层抽象--生产者和消息队列之间, 加入了互换机 (Exchange)。这样生产者和消息队列之间就没有直接联系了，转而变成生产者把消息发给互换机，互换机根据路由规则，将消息转发给指定消息队列，然后消费者从消息队列中获取消息。

基本概念

Producer(消息生产者)：向消息队列发布消息的客户端应用程序。
Consumer(消息消费者)：从消息队列获取消息的客户端应用程序。
Channel(信道)：多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接上的虚拟连接。RabbitMQ中，生产者通过和互换机建立信道来发送消息，同样消费者也需要通过和队列建立信道来获取消息。对于操作体系来说建立和销毁TCP连接都是比较昂贵的开销，所以引入了信道的概念，以复用一条TCP连接。也就说，一个TCP 被多个线程共享，每个线程对应一个信道，每个信道都有唯一的ID，包管了信道的私有性。
Message(消息)：消息由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包罗RoutingKey(路由键)、priority(消息优先权)、delivery-mode(是否持久性存储)等。
Queue(消息队列)：存储消息的一种数据结构，用来生存消息，直到消息发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者将它取走。需要注意的，当多个消费者订阅同一个队列时，该队列中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，即每一条消息只能被一个订阅者接收。
RoutingKey(路由键)：消息头的一个属性，用于标记消息的路由规则，决定了互换机的转发路径。最大长度255 字节。如下图，当我们创建好互换机和队列后，需要使用路由键将两者进行绑定，所以路由键也叫绑定键(BindingKey）。当消息生产者向互换机发送消息时，必须指定一个路由键，当互换机收到这条消息之后，会解析并获取路由键，然后同互换机和队列的绑定规则，并将消息分发到符合规则的队列中。
路由键是一个点分字符串，比如``task，quick.orange，quick.orange.rabbit`，被点号“. ”分隔开的每一段独立的字符串称为一个单词

如果互换机没有绑定任何队列，或者消息无法路由到队列，则将消息返回给生产者或者直接丢弃收到的消息，这取决于互换机的 mandatory 属性设置：
- 当 mandatory 为 true 时：如果互换机无法根据自身范例和路由键找到一个符合条件的队列，则会将该消息返回给生产者；
- 当 mandatory 为 false 时：如果互换机无法根据自身范例和路由键找到一个符合条件的队列，则会直接丢弃该消息。
Exchange(互换机)：消息的分发中央，负责接收生产者的消息，并根据肯定的路由规则将这些消息分发到差别的队列中。互换机用于转发消息，它不会存储消息。
RabbitMQ提供了四种范例的互换机：直接互换机（direct），扇出互换机（fanout），主题互换机（topic），消息头互换机（headers）
Broker(署理)：RabbitMQ服务器节点、服务器实例
Virtual Host(虚拟主机)
一个Broker中可以有多个虚拟主机，每个虚拟主机都有一套自己的互换机和队列，同一个虚拟主机中的队列和互换机不能重名，而差别的虚拟主机中的互换机和队列可同名。这样，差别的用户在访问同一个RabbitMQ Broker时，可以创建自己单独的虚拟主机，然后在自己的虚拟主机中创建互换机和队列，很好地做到了差别用户之间相互隔离的效果。

互换机范例

直接互换机（direct）

direct是互换机默认范例。
消息转发规则：当消息的路由键完全匹配队列绑定到互换机配置的路由键时，才会将消息转发到该队列。
注意，RabbitMQ默认提供了一个名称为空字符的互换机(web管理界面中名称显示为：(AMQP default))，范例为direct，绑定到所有的队列（每个队列和这个默认互换机之间的绑定键为队列自身名字），所以，有时候我们感觉不指定互换机也可以发送和接收消息，但是实际上是使用了RabbitMQ默认提供的互换机。
示例1：
如图，假设生产发送消息到直接互换机，消息路由键为：green，那么消息将被投放入Queue1队列中，因为Queue1的绑定建和消息路由键精确匹配。

示例2：
如图，假设生产发送消息到直接互换机，消息路由键为：green，那么消息将被投放入Queue1和Queue2队列中，因为Queue1，Queue2的绑定建和消息路由键精确匹配。

参考连接：
https://www.rabbitmq.com/tutorials/tutorial-four-python#direct-exchange
https://www.rabbitmq.com/tutorials/amqp-concepts#exchange-direct
https://www.rabbitmq.com/tutorials/amqp-concepts#exchange-default
扇出互换机（fanout）

消息转发规则：把所有发送到该互换机的消息转发到与该互换机绑定的所有消息队列中，与路由键，绑定键无关。
雷同于子网广播，子网内的每台主机都会获得一份复制的消息。扇出互换机转发消息是最快的。
示例：
如图，假设生产发送消息到直接互换机，消息路由键为：quick，那么最后消息将被投放入所有队列中(和路由键，绑定键无关）

参考连接：
https://www.rabbitmq.com/tutorials/tutorial-three-python
https://www.rabbitmq.com/tutorials/amqp-concepts#exchange-fanout
主题互换机（topic）

消息转发规则：当消息的路由键正则匹配队列绑定到互换机配置的路由键时，才会将消息转发到该队列。
绑定键中可以存在两种特别字符*与#，用于正则模糊匹配：

* 只能匹配一个单词
# 可以匹配0个或者多个单词。注意：当队列绑定键设置为#时，将接收所有来自互换机的消息，忽略路由键，就像fanout互换机一样。当绑定键中不使用*和#时，topic互换机和direct互换机使用效果一样。

示例：

如图，假设

生产发送消息到主题互换机，消息路由键为：quick.orange.rabbit、lazy.orange.element，那么消息将被投放入所有队列中。
生产发送消息到主题互换机，消息路由键为：quick.orange.fox，那么消息将被投放入Queue1队列中。
生产发送消息到主题互换机，消息路由键为： lazy.pink.rabbit"，那么消息将被投放入Queue2队列中，且只会放入一次，虽然匹配两个绑定键。
生产发送消息到主题互换机，消息路由键为：quick.brown.fox、 orange、quick.orange.new.rabbit，那么消息将不会被投放入任何队列中。
生产发送消息到主题互换机，消息路由键为：lazy.orange.new.rabbit，那么消息将被投放入Queue2队列中。

参考链接：
https://www.rabbitmq.com/tutorials/tutorial-five-python#topic-exchange
https://www.rabbitmq.com/tutorials/amqp-concepts#exchange-topic
消息头互换机（headers）

消息转发规则：根据消息的自界说消息头属性(headers)进行匹配路由。和路由键与绑定键无关。具体规则如下：

如果绑定队列和互换机时指定一组键值对，且此中包含一个键为x-match，键值为any或all:
- all 表现消息携带的键值对(即消息头)要全部匹配消息和队列绑定时配置的全部键值对(消息携带的键值对可以是绑定键值中没有的键值对)，才会将消息转发到该队列中。
- any 表现消息携带的键值对匹配消息和队列绑定时配置的任一键值对，就会将消息转发到该队列中。
如果在绑定队列和互换机时指定一组键值对，但是没指定x-match键值对，则默认x-match为all
如果绑定队列和互换机时未指定键值对，则互换机也会把消息发送该队列

注意：对于any和all，以字符串x-开头的消息头将不会用于路由匹配。将x-match设置为any-with-x或all-with-x，将使用以字符串x-开头的请求头进行路由匹配
参考链接：https://www.rabbitmq.com/tutorials/amqp-concepts#exchange-headers
参考链接

RabbitMQ 之使用Web管理界面认识RabbitMQ
别的工作机制

消息确认(Message acknowledgment)

消费者应用程序，即接收和处理消息的程序，大概偶然会处理消息失败，还没处理完就突然宕机，失去服务器连接，或者是其他方面的问题。这便带来了一个问题：broker应该在什么时候删除队列中的消息？AMQP 0-9-1规范赋予消费者对此的控制权。有两种确认模式：

在broker向应用程序发送消息后（使用basic.deliver或basic.get-ok方法）。
应用程序返回一个确认后（使用basic.ack方法）。

前者称为自动确认模式，而后者则称为显式确认模式。通过显式模式，应用程序可以选择何时发送确认。可以在收到消息后立即执行，也可以在处理前将其持久化到数据存储后执行，或者在完全处理消息后执行（例如，成功获取网页，处理并将其存储到某个持久化数据存储中）。需要注意的是，如果使用后者，如果应用程序忘记返回确认的话，大概会导致队列中大量消息堆积。
如果一个消费者在没有发送确认的环境下消亡，broker将把它重新传递给另一个消费者，或者如果其时没有可用的消费者，broker将等待直到至少有一个消费者注册到同一队列，然后再实验重新传递消息
参考链接：https://www.rabbitmq.com/tutorials/amqp-concepts#consumer-acknowledgements
生产者确认(producer confirmation)

使用RabbitMQ时，消息生产者发送消息后，消息是否正确到达服务器？RabbitMQ为这个问题提供了两种解决方案：

事务机制
发送者确认（发布者确认）机制

事务机制

RabbitMQ 客户端提供了三个事务机制相关的方法:

func (ch *Channel) Tx() error
func (ch *Channel) TxCommit() error
func (ch *Channel) TxRollback() error

复制代码

channel.Tx用于将当前信道设置为事务模式，channel.TxCommit用于提交事务,channel.TxRollback用于回滚交易。
通过channel.Tx打开事务后。我们可以向RabbitMQ发布消息。如果事务提交成功，则消息肯定已到达RabbitMQ。如果在事务提交和执行之前，由于RabbitMQ的异常崩溃或其他原因引发异常，则此时我们可以捕获该异常，然后通过执行channel.txRollback方法实现事务回滚。
如果想发送多条消息，只需将方法如channel.Publish 和channel.TxCommit等封装到循环里。
事务确实可以解决消息发送者和RabbitMQ之间的消息确认问题。只有当RabbitMQ成功接收到消息时，事务才能成功提交，否则，在捕获异常后可以回滚事务，同时可以重新发送消息。但是使用事务机制会严重影响RabbitMQ的性能，那么有没有更好的方法呢？RabbitMQ提供了一种改进模式，即发送方确认机制。
发送者确认机制(sender confirmation mechanism)

生产者可以将信道设置为确认(confirm)模式。一旦信道进入确认模式，信道上发布的所有消息都将被分配一个唯一的ID（从1开始）。当消息被传递到所有匹配的队列时，RabbitMQ将向生产者发送确认(Ack)（包罗消息ID），这样生产者便知道消息已正确到达。如果消息和队列是持久化范例的，则在消息写入磁盘后发送确认消息。RabbitMQ给生产者发送的确认消息中的deliveryTag包含确认消息的序列号。此外，RabbitMQ还可以在Ack方法中设置multiple参数，以指示此序列号之前的所有消息都已被处理。
事务机制中，生产者在消息发送后将进入壅闭发状态，等待RabbitMQ的响应，只有获取到响应后才可以继续发送下一条消息。相比之下，发送方确认机制是异步的。消息发布后，生产者应用程序可以在等待信道返回确认的同时继续发送下一条消息。当消息终极被确认时，生产者应用程序可以通过回调方法处理确认消息。如果RabbitMQ由于自身内部错误导致消息丢失，它将发送一个Nack指令，生产者应用程序也可以在回调方法中处理Nack指令。
在确认模式下，所有发送的消息只会被Ack或Nack一次，并且不会有消息同时被Ack和Nack的环境，但RabbitMQ不包管消息的确认速率。
生产者确认机制的缺点是无法回滚，一旦服务器崩溃，生产者无法得到确认信息，生产者本身其实也不知道该消息是否已经被持久化，只有继续重发来包管消息不丢失，但是如果原先已经持久化的消息，并不会被回滚，这样队列中就会存在两条雷同的消息，体系需要支持去重。
消息持久化(Message durability)

在AMQP 0-9-1中，队列可以声明为持久或暂时的。持久队列的元数据存储在磁盘上，而暂时队列的元数据尽大概存储在内存中。
在发布时对消息也进行了同样的区分。
默认环境下，消息不是持久的，当RabbitMQ退出或崩溃时，它会忘记队列和消息，除非你告诉它不要这样做。需要做两件事来确保消息不会丢失：需要将队列和消息都标记为持久。
起首，我们需要确保队列在RabbitMQ节点重启后能够存活。为了做到这一点，需要声明队列是持久的：

channel.queue_declare(queue='hello', durable=True)

复制代码

此时，即使RabbitMQ重新启动，hello队列也不会丢失。注意：RabbitMQ不答应使用差别的参数重新界说现有队列。
然后，通过提供一个pika.DeliveryMode.Persistent的delivery_mode属性值，将消息标记为持久消息。

channel.basic_publish(exchange='',
routing_key="task_queue",
body=message,
properties=pika.BasicProperties(
delivery_mode = pika.DeliveryMode.Persistent
))

复制代码

注意：
将消息标记为持久并不能完全包管消息不会丢失。尽管它告诉RabbitMQ将消息生存到磁盘，但照旧存在RabbitMQ接受消息但尚未生存消息的短时间窗口。此外，RabbitMQ不会对每条消息执行fsync(2)--它大概只是生存到缓存中，而不是真正写入磁盘。持久性包管性并不强，但对于我们的简单任务队列来说已经充足了。如果需要更强有力的包管，那么你可以使用发布确认。
工作队列消息分发机制

一般环境下生产者往队列里插入数据时速率是比较快的，但是消费者消费数据往往涉及到一些业务逻辑处理导致消费速率跟不上消息生产速率。因此如果一个生产者对应一个消费者的话，很容易导致很多消息堆积在队列里。这时，就得使用工作队列了--一个队列有多个消费者同时消费数据。
工作队列有两种分发数据的方式：轮询分发（round robin dispatching ）和公平分发（Fair dispatching）。
轮询分发(Round-robin dispatching)

默认环境下，RabbitMQ将按顺序将每条消息发送给下一个消费者。平均而言，每个消费者都会收到雷同数量的消息。这种分发消息的方式称为轮询(round-robin)。
但是这种分发方式存在着一些隐患，消费者虽然得到了消息，但是如果消费者没能成功处理业务逻辑，在RabbitMQ中也不存在这条消息。就会出现消息丢失并且业务逻辑没能成功处理的环境。
参考链接：https://www.rabbitmq.com/tutorials/tutorial-two-python#round-robin-dispatching
公平分发(Fair dispatch)

消费者设置每次从队列里取一条数据，并且关闭自动复兴机制，每次取完一条数据后，手动复兴并继续取下一条数据。与轮询分发差别的是，当每个消费都设置了每次只会从队列取一条数据时，并且关闭自动应答，在每次处理完数据后手动给队列发送确认收到数据。
RabbitMQ只是在消息进入队列时分派消息。它不思量消费者未确认的消息数量。它只是盲目地将每第n条消息发送给第n个消费者。例如，在有两个worker的环境下，当奇数消息量比较大，偶数消息量比较少时，一个worker会一直很忙，另一个几乎不做任何工作。但是RabbitMQ对此一无所知，仍然会均匀地发送消息。
为相识决这个问题，可以使用Channel#basic_qos信道方法，并将prefetch_count设置为1。这使用basic.qos协议方法告诉RabbitMQ不要一次给一个worker多条消息。或者，换句话说，在处理并确认前一条消息之前(要求关闭自动确认)，不要向worker发送新消息。相反，它会将其发送给下一个不忙的worker。这样队列就会公平给每个消息费者发送数据了。

channel.basic_qos(prefetch_count=1)

复制代码

注意队列大小
如果所有的worker都很忙，你的队列大概会排满。需要密切关注这一点，大概可以添加更多的worker，或者使用消息TTL。

参考链接：https://www.rabbitmq.com/tutorials/tutorial-two-python#fair-dispatch
备用互换机(Alternate Exchange）

有时希望让客户端处理互换无法路由的消息（即，要么是因为没有绑定队列，要么是没有匹配的绑定）。典范的例子

检测客户端不测或恶意发布无法路由的消息
或者路由语义，此中一些消息被特别处理，其余消息由通用处理程序处理

备用互换机(简称AE）就是是一个解决这些场景的功能。
当发布到具有配置AE的互换机的消息无法路由到任何队列时，信道会将消息重新发布到指定的AE。如果该AE不存在，则天生警告日志。如果AE无法路由消息，则它会将消息发布到其AE（如果已配置）。此过程会一直持续到消息成功路由、到达AE链的末尾，或者遇到已经实验路由消息的AE。
例如，如果使用路由键key1将消息发布到my-direct，则该消息将根据标准AMQP行为路由到routed队列。但是，当使用路由键key2将消息发布到my direct时，消息不会被丢弃，而是通过我们配置的AE路由到unrouted队列。
AE的行为纯粹与路由有关。如果消息通过AE路由，则出于mandatory标记的目的，它仍被视为已路由的，否则消息不变。
参考链接：https://www.rabbitmq.com/docs/ae
队列长度限定(Queue Length Limit)

你可以设置队列的最大长度。最大长度限定可以设置为消息数，也可以设置为字节数(所有消息体长度的总和，忽略消息属性和任何开销），或同时设置两者)
要设置最大长度（任何一种范例），你可以使用策略（强烈发起使用此选择）或由客户端使用队列的可选参数进行界说。同时使用策略和参数两种方式界说最大长度的环境下，则使用指定的两个值中的最小值。
使用operator策略配置的队列长度设置。
在所有环境下，都使用处于就绪状态的消息数量。消费者未确认的消息不计入限定。
使用rabbitmqctl list_queues输出中的messages_ready和message_bytes_ready，以及管理UI和HTTP API响应中雷同命名的字段，可以观察到就绪态消息的数量及其字节占用空间。
默认最大队列长度限定行为

当设置了最大队列长度或大小并且达到最大值时，RabbitMQ的默认行为是从队列前面丢弃或死信消息（即队列中最旧的消息）。要修改此行为，请使用下面描述的溢出设置。
队列溢出行为

使用overflow设置配置队列溢出行为。如果将overflow设置为reject-publish 或 reject-publish-dlx，则最近发布的消息将被丢弃。此外，如果启用了发布者确认，则将通过basic.nack消息通知发布者拒绝。如果消息被路由到多个队列并被此中至少一个队列拒绝，则信道将通过basic.nack通知发布者。消息仍将发布到所有其他可以列队的队列中。reject-publish 和 reject-publish-dlx之间的区别在于，reject-publish-dlx还会｀dead-letter`被拒绝消息。
参考链接：
Queue Length Limit | RabbitMQ
死信互换机(Dead letter exchange)

什么是死信互换机
来自队列的消息可以是“死信”的，这意味着当发生以下任意一个事件时，这些消息会被重新发布到互换机。

消费者使用basic.reject或basic.nack对消息进行否定确认，并将requeue参数设置为false，或
由于每条消息的TTL，消息逾期，或
因为其队列凌驾了长度限定，消息被丢弃
或消息返回到仲裁队列(quorum queue)的次数凌驾了delivery-limit。

如果整个队列逾期，则队列中的消息不会成为死信。
死信互换（DLX）也是正常的互换。它们可以是任何常见的范例，并被声明为正常范例。
对于任何给定的队列，DLX可以由客户端使用队列的参数界说，也可以在服务器中使用策略界说。如果策略和参数都指定了DLX，则参数中指定的DLX将覆盖策略中指定的那个DLX。
发起使用策略进行配置，因为它答应DLX重新配置，而不涉及应用程序重新部署。
使用策略配置死信互换机

如需使用策略指定DLX，则将dead-letter-exchange添加到策略界说中。例如：
rabbitmqctlrabbitmqctl set_policy DLX ".*" '{"dead-letter-exchange":"my-dlx"}' --apply-to queuesrabbitmqctl (Windows)rabbitmqctl set_policy DLX ".*" "{""dead-letter-exchange"":""my-dlx""}" --apply-to queues上述策略将my-dlx死信互换机应用于所有队列。这只是一个例子，在实践中，差别的队列集通常使用差别的死信设置（或根本不使用）。
雷同的，可以通过将dead-letter-routing-key添加到策略中来指定显式路由键。
使用可选队列参数配置死信互换机

要为队列设置DLX，在声明队列时指定可选参数x-dead-letter-exchange 。该参数值必须是同一虚拟主机中的互换机名称：

channel.exchangeDeclare("some.exchange.name", "direct");
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange", "some.exchange.name");
channel.queueDeclare("myqueue", false, false, false, args);

复制代码

上述代码声明了一个名为some.exchange.name的新互换机，并将此新互换机设置为新创建队列的死信互换机。注意，在声明队列时不必声明互换机，但在消息变成死信时，它必须已经存在。如果缺失了，则消息会自动被丢弃。
还可以指定一个路由键，以便在消息成为死信时使用。如果未设置路由键，则使用消息自己的路由键。

args.put("x-dead-letter-routing-key", "some-routing-key");

复制代码

死信路由

死信消息会被路由到死信互换机：

使用为他们所在的队列指定的路由键；或者如果没有设置，
使用与最初发布时雷同的路由键

例如，如果你用foo路由键将一条消息发布到一个死信互换机，且该消息是死信，那么它就会用foo路由键发布到死信互换机。如果消息最初到达的队列声明为x-dead-letter-route-key设置为bar，则消息将使用bar路由键发布到其死信互换。
注意，如果没有为队列设置特定的路由键，则队列上的消息将使用所有原始路由键进行死信处理。这包罗由·CC和BCC`头添加的路由键。
形成一个消息死信循环是大概的。例如，当将“死信”消息列队到默认互换机而不指定死信路由键时，就会发生这种环境。如果在整个循环中没有拒绝，则此类循环中的消息（即两次到达同一队列的消息）将被丢弃。
参考链接：
Dead Letter Exchanges | RabbitMQ
优先级队列(Priority Queue)

什么是优先级队列

RabbitMQ支持向经典队列(classic queue)加priorities。打开priority功能的经典队列通常被称为“优先级队列”。优先级支持 1 到 255，但强烈发起使用 1 到 5 之间的值。重要的是要知道，更高的优先级值需要更多的CPU和内存资源，因为RabbitMQ需要在内部为从1到为给定队列配置的最大值的每个优先级维护一个子队列。
通过使用客户端提供的可选参数x-max-priority，经典队列可以成为优先级队列。
使用客户端提供的可选参数

要声明优先级队列，使用x-max-priority可选队列参数。此参数应为1到255之间的正整数，表现队列应支持的最大优先级。例如，使用Java客户端：

Channel ch = ...;
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-max-priority", 10);
ch.queueDeclare("my-priority-queue", true, false, false, args);

复制代码

然后，发布者可以使用basic.properties的优先级字段发布优先消息。数字越大表现优先级越高。
设计不支持使用策略将经典队列声明为优先级队列。
优先级队列行为

AMQP 0-9-1规范对优先级的预期工作方式有点模糊。它规定所有队列必须支持至少2个优先级，最多支持10个。它没有界说如何处理没有优先级属性的消息。
默认环境下，RabbitMQ经典队列不支持优先级。创建优先级队列时，可以根据需要选择最大优先级。在选择优先级值时，需要思量以下因素：

每个队列的每个优先级都有一些内存和磁盘成本。还有一个额外的CPU成本，特别是在消费时，所以你大概不希望创建大量的级别。
消息优先级字段被界说为无符号字节，因此在实践中优先级应在0到255之间。
没有设置优先级属性的消息被视为优先级为0。优先级高于队列最大值的消息将被视为以最大优先级发布。

最大优先级和资源占用

如果优先级队列是你想要的，强烈发起使用之前声明的介于1和5之间的值。如果必须高于5，则1到10之间的值就充足了（保持为个位数），因为当前使用更多的优先级会通过使用更多的Erlang进程从而消耗更多的CPU资源。运行时调治也会受到影响。
优先级队列如何与消费者协同工作

如果消费者连接到随后将发布消息的空优先级队列，则在消费者接受这些消息之前，消息大概不会在优先级队列中等待任何时间（所有消息都会立即被接受）。在这种环境下，优先级队列没有任何机会对消息进行优先级排序，因此不需要优先级。
然而，在大多数环境下，前面所述场景并不常见，因此应该在消费者手动确认模式下使用basic.qos(预取)方法来限定任何时候可以发送的消息数量，并答应对消息进行优先级排序。basic.qos是消费者在连接到队列时设置的值。它表现消费者一次可以处理多少条消息。
以下示例试图更详细地解释消费者如何使用优先级队列，并强调有时当优先级队列与消费者一起使用时，优先级较高的消息实际上大概需要等待优先级较低的消息起首被处理。
示例

一个新的消费者连接到一个空的经典（非优先级）队列，消费者预取（basic.qos）为10。
消息被发布并立即发送给消费者进行处理。
然后，又有5条消息被快速发布并立即发送给消费者，因为消费者声明的预取数量为10条消息，现在只取到1条。
接下来，又有10条消息被快速发布并发送给消费者，10条消息中只有4条被发送给消费者（因为已达到最初的basic.qos值10），其余6条消息必须在队列中等待（就绪态消息）。
消费者现在确认了5条消息，因此现在上面等待的6条消息中的5条被发送给消费者。
现在添加优先级
如上例所示，消费者连接队列时使用basic.qos，并设置值为10。
10条低优先级消息被发布并立即发送给消费者（现已达到basic.qos限定）
发布一条最高优先级的消息，但现在已凌驾预取值，因此最高优先级消息需要等待优先级较低的消息先被处理。

与其他功能的交互

一般来说，优先级队列具有标准RabbitMQ队列的所有功能：支持持久性、分页、镜像等。不外，照旧有几个交互需要开发人员注意的

应该逾期的消息仍然只从队列的开头逾期。这意味着，与普通队列差别，即使是每个队列的TTL也大概导致逾期的低优先级消息卡在未逾期的高优先级消息背面。这些消息永远不会被传递，但它们会出现在队列统计信息中。
设置了最大长度的队列会像往常一样从队列头部删除消息以强制执行限定。这意味着较高优先级的消息大概会被丢弃，为较低优先级的消息让路，这大概不是你所期望的。

为什么策略界说不支持优先级

为队列界说可选参数最方便的方法是使用策略。策略是配置TTL、队列长度限定和其他可选队列参数的推荐方法。
但是，策略不能用于配置优先级，因为策略是动态的，可以在声明队列后更改。优先级队列在队列声明后永远无法更改其支持的优先级数量，因此使用策略不是一个安全的选择。
参考链接：
Classic Queues Support Priorities | RabbitMQ
耽误消息(Delaying Messages)

一段时间以来，人们一直在寻找使用RabbitMQ实现耽误消息传递的方法。到现在为止，公认的解决方案是使用James Carr在这里提出的消息TTL和死信互换机的组合。RabbitMQ为此提供了一个开箱即用的解决方案--RabbitMQ耽误消息插件(rabbitmq_delayed_message_exchange)
RabbitMQ耽误消息插件为RabbitMQ添加了一种新的互换机范例，路由到该互换机的消息可以被耽误，如果用户选择这样做的话。
要耽误消息，用户必须使用名为x-delay的特别消息头发布消息，x-delay为一个整数，表现RabbitMQ应耽误消息的毫秒数。值得注意的是，这里的耽误意味着：耽误消息路由到队列或其他互换机。互换机没有消费者的概念。因此，一旦耽误到期，插件将实验将消息路由到与互换机的路由规则相匹配的队列。请注意，如果消息无法路由到任何队列，那么它将被丢弃，正如AMQP对不可路由消息所指定的那样。

// ... elided code ...
byte[] messageBodyBytes = "delayed payload".getBytes();
AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder();
headers = new HashMap<String, Object>();
headers.put("x-delay", 5000);
props.headers(headers);
channel.basicPublish("my-exchange", "", props.build(), messageBodyBytes);

复制代码

参考链接：
https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq#delaying-messages
生存时间功能(Time-To-Live Feature)

使用RabbitMQ，你可以为消息和队列设置TTL(time-to-live)参数或策略。顾名思义，TTL指定了消息和队列“存活”的时间段。
消息的TTL决定消息在队列中可以保留多长时间。如果队列中消息的保留期凌驾队列的消息TTL，则消息逾期并被丢弃。
“丢弃”意味着消息将不会传递给任何订阅了该队列的消费者，也无法通过basic.get方法访问获取。消息TTL可以应用于单个队列、一组队列，也可以逐个应用于消息。
TTL也可以在队列上设置，而不仅仅是队列内容。此功能可以与auto-delete队列属性一起使用。在队列上设置TTL通常只对非持久性Classic队列有意义。流不支持逾期。
队列只有在不使用时才会在一段时间后逾期（如果队列有在线消费者，则表现队列正在使用中）。
TTL行为由队列optional参数控制，配置它的最佳方法是使用策略。
TTL设置也可以由operator策略强制执行。
队列中每个队列消息TTL(Per-Queue Message TTL in Queues)

通过使用策略设置Message-TTL参数，可以为给定的队列设置每条消息的TTL或者在队列声明时指定雷同的参数。
在队列中停留时间凌驾TTL配置的消息称为“逾期”消息。请注意，路由到多个队列的消息在其驻留的每个队列中的逾期时间大概不一样，或者根本不会逾期。一个队列中消息的死亡不会影响其他队列中同一消息的生存期。
服务器包管不会使用basic.devely传递逾期的消息给消费者，也不会作为对消费者轮询的响应（basic.get-ok响应）发送。
此外，服务器将实验在基于TTL的到期时或到期后不久删除消息。
TTL参数或策略的值必须为非负整数（等于或大于零），以毫秒为单位描述TTL周期。
因此，值1000意味着添加到队列中的消息将在队列中停留1秒，或者直到它被传递给消费者。参数可以是AMQP 0-9-1范例的short-short int、short int、long int或long-long int。
使用策略为队列界说消息TTL

要使用策略指定TTL，将键message-TTL添加到策略界说中：
rabbitmqctlrabbitmqctl set_policy TTL ".*" '{"message-ttl":60000}' --apply-to queuesrabbitmqctl (Windows)rabbitmqctl set_policy TTL ".*" "{""message-ttl"":60000}" --apply-to queues这将对所有队列应用60秒的TTL。
声明队列时使用x-arguments为队列界说消息TTL

Java中的这个例子创建了一个队列，消息最多可以驻留60秒：

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 60000);
channel.queueDeclare("myqueue", false, false, false, args);

复制代码

可以将消息TTL策略应用于已包含消息的队列，但这涉及一些注意事项。
如果消息被重新列队（例如，由于使用了具有重新列队参数的AMQP方法，或由于信道关闭），则保留消息的原始逾期时间。
将TTL设置为0会导致消息在到达队列时逾期，除非它们可以立即传递给消费者。因此，这为RabbitMQ服务器不支持的 immediate发布标记提供了一种替代方案。与 immediate标记差别，不会发起basic.return，如果设置了死信互换机，则消息将变为死信。
消息发布者中每条消息的TTL(Per-Message TTL in Publishers)

通过设置expiration属性，发布消息时可以基于每条消息指定TTL。
expiration 字段的值以毫秒为单位描述TTL周期。与x-message-ttl具有雷同的束缚。由于expiration 字段必须是字符串，因此署理将（仅）接受该数字的字符串表现。
当同时指定了每队列( per-queue)和每消息(per-message)TTL时，将选择两者之间的较小值。
示例：使用RabbitMQ Java客户端发布一条在队列中最多驻留60秒的消息：

byte[] messageBodyBytes = "Hello, world!".getBytes();
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
.expiration("60000")
.build();
channel.basicPublish("my-exchange", "routing-key", properties, messageBodyBytes);

复制代码

每条消息的TTL和死信互换(Per-Message TTL and Dead Lettering)

Quorum Queues

当逾期的消息到达队列头部时，仲裁队列将其设置为死信(原文：Quorum queues dead letter expired messages when they reach the head of the queue.)
Classic Queues

在某些环境下，经典队列会将逾期消息设置为死信（原文：Classic queues dead letter expired messages in a few cases）:

当消息到达队列的头部时
当队列收到影响它的策略更改的通知时

追溯应用Per-message TTL到现有队列(Per-message TTL Applied Retroactively (to an Existing Queue))

追溯应用每条消息的TTL到现有队列。
被应用了Per-message TTL的队列（当它们已经有消息时）将在发生特定事件时丢弃这些消息。
只有当逾期的消息到达队列的头部时，它们才会被真正丢弃（标记为删除）。消费者将不会收到逾期的消息。请记住，消息逾期和消费者传递之间大概存在自然的竞争条件，例如，消息可以在写入socket后，到达消费者之前逾期。
当设置了每条消息TTL时，逾期消息可以排在未逾期消息之后，直到后者被消耗或逾期。因此，这些逾期消息使用的资源将不会被释放，它们将被计入队列统计数据中（例如队列中的消息数量）。
当追溯应用每消息TTL策略时，发起让消费者在线，以确保更快地丢弃消息。
鉴于现有队列上设置每条消息TTL的这种行为，当需要删除消息以释放资源时，应使用队列TTL（或队列清除，或队列删除）。
队列TTL(Queue TTL)

TTL也可以在队列上设置，而不仅仅是队列内容。此功能可以与auto-delete队列属性一起使用。
在队列上设置TTL（逾期）通常只对瞬态（非持久）经典队列有意义。流不支持逾期。
队列只有在不使用时才会在一段时间后逾期（如果队列有在线消费者，则表现队列正被使用）。
可以通过将x-expires参数设置为queue.restable，或通过设置expires策略来为给定的队列设置逾期时间。这可控制队列在被自动删除之前可以使用多长时间。未使用意味着队列没有消费者，队列最近没有重新声明（重新声明会续订租约），并且至少在到期期间没有basic.get调用。
服务器包管，如果队列至少在逾期期间未使用，则将被删除。但无法包管如何迅速移除逾期后队列。
x-expires参数或expires策略值描述了以毫秒为单位的逾期时间。逾期时间必须是正整数（与消息TTL差别，它不能为0）。因此，值1000表现将删除1秒内未使用的队列。
使用策略界说队列TTL(Define Queue TTL for Queues Using a Policy)

以下策略使所有队列在上次使用后30分钟后逾期：
rabbitmqctlrabbitmqctl set_policy expiry ".*" '{"expires":1800000}' --apply-to queuesrabbitmqctl (Windows)rabbitmqctl.bat set_policy expiry ".*" "{""expires"":1800000}" --apply-to queues声明队列时使用x-arguments为队列界说队列TTL(Define Queue TTL for Queues Using x-arguments During Declaration)

此Java示例创建了一个队列，该队列在未使用30分钟后逾期。

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-expires", 1800000);
channel.queueDeclare("myqueue", false, false, false, args);

复制代码

参考链接：
Time-To-Live and Expiration | RabbitMQ

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)