跳至主内容

RabbitMQ 4.0：新的 Quorum Queue 功能

·阅读 12 分钟
Michał Kuratczyk
Michał Kuratczyk

RabbitMQ 4.0（目前为 beta 版）包含新的 Quorum Queue 功能

  • 消息优先级
  • 消费者优先级与单活跃消费者结合
  • 默认投递限制现为 20（破坏性更改！）
  • 长队列的更快恢复

消息优先级​

消息优先级支持可能是仲裁队列最受期待的功能，主要由现有经典镜像队列用户请求迁移到仲裁队列（请记住，经典队列镜像支持已在 4.0 中移除）。

然而，优先级支持的方式与经典队列处理优先级的方式有很大不同。经典队列需要 x-max-priority 参数来定义给定队列的最大优先级数量（如果未提供此参数，队列将平等对待所有消息）。技术上允许 0 到 255 的值，但实际上每个队列最多应使用少量优先级。仲裁队列不需要任何预先声明（无需为给定队列启用优先级），但每个队列只有两种优先级：普通和高。其行为与 AMQP 1.0 规范一致（参见AMQP 1.0 规范第 3.2.1 章）。

  • 优先级值在 0 到 4 之间（含）被视为普通优先级。
  • 高于 4 的任何值都被视为高优先级。
  • 如果发布者未指定消息的优先级，则假定值为 4（普通优先级）。

如果仲裁队列同时包含普通优先级和高优先级消息，消费者将同时接收两者的混合，比例为每 1 条普通优先级消息对应 2 条高优先级消息。这种方法可以避免饿死，因为无论高优先级消息的数量多少，普通优先级消息的处理也会取得进展。这与经典队列实现不同，经典队列总是先传递高优先级消息（如果存在），因此普通优先级消息可能永远不会被传递（或者，更可能的是，其传递延迟会非常高）。

这是如何工作的可视化表示。为了进行此测试，我们首先发布了 100k 条普通优先级消息，然后发布了 100k 条高优先级消息。由于在 4.0 之前仲裁队列不具备优先级感知能力，如果我们将其用于旧版本，然后启动一个消费者，它将首先接收普通优先级消息（因为它们更早），然后是所有高优先级消息。使用 4.0，我们可以看到消费者立即开始接收混合消息，每秒约 1500 条普通优先级消息和两倍数量的高优先级消息，总计约 4500 条消息/秒（实际传递速率并不重要，它们取决于许多因素；2:1 的高/普通优先级比例是优先级上下文中的关键）。一旦队列传递完所有高优先级消息，消费者将开始接收每秒约 4500 条普通优先级消息——只要它在此测试场景中能够处理的数量。蓝色的虚线（带有右侧的轴刻度）是队列中准备就绪的消息数量（两种优先级总计）——我们可以看到它从 200k 开始，最终降至零。

Consumer receives a mix of high (yellow) and normal priority (green) messages
消费者接收高优先级（黄色）和普通优先级（绿色）消息的混合

让我们考虑相反的情况——如果我们先发布所有高优先级消息，然后再发布所有普通优先级消息怎么办？在这种情况下，消费者将按发布顺序接收消息。普通优先级消息没有任何理由可以超越更高优先级的消息。

Normal priority messages (green) don't overtake high priority (yellow) messages published first
先发布的高优先级消息（黄色）不会被普通优先级消息（绿色）超越

如何进行此测试？

对于此测试，我们使用了 omq，这是一个 AMQP 1.0、MQTT 和 STOMP 的测试客户端。仲裁队列的行为不取决于使用的协议——AMQP 1.0 只是被使用，因为 omq 会按消息优先级发出消息消费指标。

# declare a quorum queue (you can use the Management UI or any other method)
rabbitmqadmin declare queue name=qq queue_type=quorum
# publish normal priority messages (10 publishers, 10k messages each)
omq amqp --publishers 10 --consumers 0 --publish-to /queues/qq --message-priority 1 --pmessages 10000
# publish high priority messages
omq amqp --publishers 10 --consumers 0 --publish-to /queues/qq --message-priority 10 --pmessages 10000
# consume all messages from the queue
omq amqp --publishers 0 --consumers 1 --consume-from /queues/qq --consumer-credits 100

对于第二种场景，只需反转发布命令的顺序即可。

如果我需要更多控制权怎么办？​

如果 2:1 的传递比例的两种优先级不符合您的要求，我们可以推荐两点：

  1. 重新审视您的需求。😄对具有多个优先级的消息传递顺序进行推理非常困难。最好确保所有消息都足够快地传递，并且优先级仅用于确保在偶尔出现长队时，重要消息可以跳过队列。
  2. 如果您确实需要更多优先级和/或对不同优先级如何处理的更多控制，使用多个队列是您的最佳选择。您可以开发一个订阅多个队列的消费者，然后决定从哪个队列进行消费。

消费者优先级与单一活跃消费者结合​

从 RabbitMQ 4.0 开始，仲裁队列将在选择单一活跃消费者时考虑消费者优先级。如果出现一个优先级更高的消费者（订阅），仲裁队列将切换到它。这特别有用，如果您有多个队列每个队列应该只有一个消费者，但您不希望应用程序的单个实例成为所有这些队列的消费者，这可能发生在第一个启动的应用程序实例订阅了所有这些单活跃消费者队列时。现在，您可以为订阅不同队列时选择不同的优先级，以确保每个实例仅从其“首选”队列进行消费，并仅作为其他队列的备用消费者。

为了更好地解释此功能，让我们回顾一下所有组成部分。 单一活跃消费者 是一个队列参数，它可以阻止队列向多个消费者传递消息，无论有多少个订阅了该队列。一个消费者是活跃的，所有其他消费者都是不活跃的。如果活跃消费者断开连接，其他消费者之一将被激活。此功能用于需要严格的消息处理顺序时。

消费者优先级 允许您指定，而不是以公平轮询方式将消息传递给所有订阅的消费者（这是经典队列和仲裁队列的默认行为），应该优先选择某个消费者。

直到 4.0 版本，这些功能在很大程度上是相互排斥的——如果启用了单一活跃消费者，只要前一个消费者保持活跃，新的消费者就永远不会变得活跃，无论其优先级如何。从 4.0 开始，如果新消费者的优先级高于当前活跃消费者的优先级，仲裁队列将切换到优先级更高的消费者：它将停止向当前消费者传递消息，等待所有消息被确认，然后将旧消费者置为非活跃状态，并激活优先级更高的消费者。

下图显示了此行为。图中有三个指标：

  • 绿色线显示第一个（默认优先级）消费者消耗的消息数量（该消费者配置为每秒消耗 10 条消息）
  • 黄色线显示第二个、优先级更高的消费者消耗的消息数量相同
  • 蓝色线显示未确认的消息数量（右侧轴刻度）

Single Active Consumer switchover: the normal-priority consumer (green) gets deactivated after it has acknowledged its messages, then the higher-priority consumer (yellow) gets activated
单一活跃消费者切换：普通优先级消费者（绿色）在确认其消息后被停用，然后高优先级消费者（黄色）被激活。

最初，我们只有一个消费者，并且正如预期的那样，它每秒消耗 9-10 条消息（这些跳跃发生在 9 和 10 之间只是指标的发出和显示方式的结果）。该消费者配置的预取值为 1000 条消息，并且由于队列中有大量消息——预取缓冲区被最大化利用。然后出现黄色线，最初为 0 条消息/秒。这是高优先级消费者，它已经连接，但尚未激活。从它连接的那一刻起，我们可以看到未确认消息的数量在下降，因为队列不再向原始消费者传递消息。一旦所有消息都被确认，新消费者就成为单一活跃消费者，并接收 1000 条消息，因为这是它的预取值。然后它将按配置愉快地每秒消耗约 10 条消息。

如何进行此测试？

对于此测试，我们使用了 perf-test，一个 AMQP 0.9.1 的测试客户端。

# Publish 5000 messages to have a backlog (perf-test will declare a quorum queue `qq-sac`)
perf-test --quorum-queue --queue qq-sac --pmessages 5000 --confirm 100 -qa x-single-active-consumer=true --consumers 0
# Start a consumer with the default priority and prefetch of 1000; consume ~10 msgs/s
perf-test --producers 0 --predeclared --queue qq-sac --consumer-latency 100000 --qos 1000
# In another window, some time after starting the first consumer, start a higher priority consumer
perf-test --producers 0 --predeclared --queue qq-sac --consumer-latency 100000 --qos 1000 --consumer-args x-priority=10

一段时间后，您应该会看到第一个消费者停止接收消息（perf-test 不再输出），而第二个消费者接收消息。

注意

此示例中使用的设置是为了突出切换过程而选择的，并不适用于实际场景。如果一个消费者每秒只能处理 10 条消息，通常没有理由将预取值配置得像 1000 那么高。

默认传递限制现为 20​

警告

这可能对某些应用程序来说是一个破坏性更改。

仲裁队列现在默认设置为 20 的传递限制。过去，没有设置限制，因此仲裁队列会一直尝试传递，直到消息被消费者确认或丢弃。这可能导致消息卡在队列中，永远无法传递。

此更改的缺点是，如果未配置死信队列，消息将在 20 次尝试后被丢弃。因此，强烈建议为所有仲裁队列配置死信队列。

更快地恢复长队列​

这与其说是一个功能，不如说是一个内部更改，但绝对值得一提。直到现在，如果 RabbitMQ 节点重启，该节点上的所有仲裁队列都必须从上次快照开始读取其所有数据（Raft 日志）以重建其内存状态。例如，如果您现在向仲裁队列发布数百万条消息，然后重启一个节点，您会发现在节点启动后，队列将在相当长一段时间内（至少几秒钟）报告 0 条待处理消息，并且您将无法开始消费这些消息。队列还没有准备好处理流量——它仍在从磁盘读取数据（注意：这并不意味着所有数据都保留在内存中，绝大多数都没有，但队列数据的索引/摘要是）。从 RabbitMQ 4.0 开始，仲裁队列创建检查点文件，其中包含队列在特定时间点的状态。启动时，队列可以读取最新的检查点以及该时间点之后 Raft 日志的部分。这意味着仲裁队列的启动时间大大缩短。

例如，在一台机器上，包含 1000 万条 12 字节消息的单个仲裁队列的 RabbitMQ 节点需要大约 30 秒才能启动。使用 RabbitMQ 4.0，只需不到一秒钟。

您可能会想，快照和检查点有什么区别？在很多方面，它们是相同的——它们实际上共享将它们写入磁盘的代码。区别在于快照仅在 Raft 日志被截断时创建。对于许多常见的队列用例，这已经足够了——旧消息已被消费，我们创建一个不再包含它们的快照，并截断日志。此时，队列不再记得这些消息曾经存在过。而检查点则在无法截断日志时定期创建。测试场景是一个很好的例子——由于我们没有消费任何消息，最旧的消息仍然存在，我们不能就这样忽略它们。但检查点仍然允许队列更快地启动。当日志被截断时（在此示例中——在消费了部分旧消息之后），检查点可以提升为快照。

我如何尝试？

我们将再次使用 perf-test 来声明队列并发布消息。

# Publish 10 million 12-byte messages (feel free to play with other values)
perf-test --quorum-queue --queue qq --consumers 0 --pmessages 5000000 --confirm 1000 --producers 2
# restart the node
rabbitmqctl stop_app && rabbitmqctl start_app
# list the queues (repeat this command until the number of messages is 10 million instead of 0)
rabbitmqctl list_queues

总结​

RabbitMQ 4.0 是 RabbitMQ 的一个重要里程碑。随着经典队列镜像的移除，仲裁队列成为高可用、复制队列的唯一选择（注意：流也是高可用和复制的，但技术上不是队列；尽管如此，对于过去使用经典镜像队列的某些用例，它们可能仍然是一个不错的选择）。多年来，仲裁队列提供了比镜像队列更高的数据安全保证和更好的性能，通过这些最新的改进，它们在更广泛的场景中变得更加健壮和高效。

您现在可以试用 RabbitMQ 4.0 beta：https://github.com/rabbitmq/rabbitmq-server/releases/tag/v4.0.0-beta.5

© . This site is unofficial and not affiliated with VMware.