异步解耦之RabbitMQ(三)

异步解耦之RabbitMQ(一)

异步解耦之RabbitMQ(二)_RabbitMQ架构及交换机

RabbitMQ提供了许多功能和选项，包括队列和消息的 TTL（Time-To-Live，生存时间）。在本篇博客中，我们将深入探讨 RabbitMQ 队列和消息的 TTL，以及如何使用它们来管理消息的过期和清理。

什么是队列?

RabbitMQ中的队列是消息的有序集合。消息以FIFO(“先进先出”)的方式进入和退出队列(传递给消费者)。

要用通用术语定义队列，它是一个具有两个主要操作的顺序数据结构:一个项目可以在尾部加入队列(添加)，并从头部退出队列(消耗)。

队列在消息传递技术领域中扮演着重要角色。许多消息传递协议和工具假设发布者和使用者使用类似队列的存储机制进行通信。

消息传递系统中的许多特性都与队列相关。一些RabbitMQ队列特性，比如消费者的优先级和排队，会影响消费者观察到的排序。

RabbitMQ 的队列类型各具特点，可以根据不同的场景和需求，结合使用这些队列类型，从而实现更灵活和高效的消息传递。消息队列是RabbitMQ的核心概念之一，通过队列来实现生产者和消费者之间的解耦，在高并发场景下保证了系统的可靠性和稳定性。

RabbitMQ普通队列

1. 简单队列模式（Simple Queue）

简单队列是RabbitMQ最基本的队列模式，它实现了一对一的消息传递。生产者将消息发送到队列，消费者从队列中接收消息。使用该模式时，只需创建一个队列和一个消费者，生产者将消息发送到队列，消费者从队列中获取消息处理。

使用场景：适用于简单的任务处理，不需要多个消费者并发处理。

缺点：

只适用于单个生产者和单个消费者的场景。
没有消息的路由规则和分发机制。

2. 工作队列模式（Work Queue）

工作队列模式适用于多个消费者共享一个队列的场景。消息被平均分发给各个消费者，每个消费者都会接收到一部分消息。可以通过增加消费者实现负载均衡。使用该模式时，多个消费者同时监听同一个队列，并从中取出消息进行处理，处理完毕后再将消息标记为已完成。

使用场景：适用于需要多个消费者并发处理消息的任务，如爬虫任务、日志处理等。

缺点：

没有消息的路由规则和分发机制。
无法根据消息的优先级进行处理。

3. 发布/订阅模式（Publish/Subscribe）

发布/订阅模式适用于消息广播给多个消费者的场景，每个消费者都有自己的队列。消息被广播到所有队列中。使用该模式时，生产者发送消息到交换机，交换机将消息分发到所有与之绑定的队列，每个队列都有一个消费者监听并处理其中的消息。

使用场景：适用于消息广播、多个消费者监听同一个事件的场景，如用户注册成功后，需要发送邮件和短信通知用户。

缺点：

无法根据消息的内容进行处理。
可能会有一些冗余的消息，因为所有的订阅者都会接收相同的消息。

4. 路由模式（Routing）

路由模式根据消息的路由键将消息发送给指定的队列，消费者只接收特定路由键的消息。使用该模式时，生产者将消息发送到交换机，并指定一个路由键，交换机根据路由键将消息发送到符合条件的队列。

使用场景：适用于需要根据消息的类型或内容将消息分发到特定队列的场景，如商品订单系统中，根据订单状态将消息发送到不同的队列进行处理。

缺点：

需要配置和管理路由规则。
性能可能会受到一定影响，因为需要进行路由操作。

5. 主题模式（Topic）

主题模式根据消息的主题标签将消息发送给匹配的队列，支持通配符匹配。使用该模式时，生产者将消息发送到交换机，并指定一个主题标签，交换机按照规则将消息发送到符合条件的队列。主题标签使用"."分隔，可以使用"*"和"#"通配符进行模糊匹配。

使用场景：适用于需要根据消息的主题标签将消息分发到特定队列的场景，如博客平台中，根据博客主题将消息分发到对应的队列进行处理。

缺点

需要对消息的路由键进行配置和管理，需要一定的维护成本。
在路由键匹配时，性能可能会受到一定的影响，需要进行额外的计算和处理。

RabbitMQ 高级队列

1. 延迟队列（Delay Queue）：

延迟队列是用于在需要延迟触发任务或实现延迟处理的应用场景中非常有用的一种队列。它允许您将消息发送到队列，并指定一个延迟时间后才能被消费者接收和处理。

使用场景：

实现定时任务：您可以使用延迟队列来触发定时任务的执行，例如定时发送邮件或生成报告。
延迟消息发送：您可以通过延迟队列来实现稍后发送消息的功能，例如发送短信验证码或延迟处理用户请求。

优点：

简单实现：借助 RabbitMQ 的插件或结合 TTL（Time-To-Live）和死信队列来实现延迟队列非常简单。
灵活性高：您可以根据需求设置不同的延迟时间，从几秒钟到几天不等。

缺点：

RabbitMQ 并不直接支持延迟队列，需要通过插件或结合其他特性来实现。
在大规模延迟队列的情况下，可能会对系统的性能和资源产生一定影响。

2. 死信队列（Dead Letter Queue）：

死信队列是用于处理消息被拒绝、超时或重新排队次数超过限制等失败情况的一种队列。当消息无法正常被消费时，它们将被发送到死信队列中，以便进行调试和分析。

使用场景：

处理异常情况：您可以利用死信队列来处理由于故障、错误或不可预见的问题导致的消息处理失败的情况。
调试和分析：通过观察死信队列中的消息，您可以更好地了解和分析发生的错误，并采取相应措施。

优点：

增强了消息处理的可靠性：即使消息在正常队列中无法被处理，也能够通过死信队列进行后续处理。
方便进行故障排查：通过查看死信队列中的消息，您可以更好地了解出现问题的原因和位置。

缺点：

需要额外的配置队列属性和消费者参数来设置死信队列的行为。
可能会增加消息处理的复杂性和额外的系统资源消耗。

5. 优先级队列（Priority Queue）：

优先级队列是用于在具有不同优先级的消息中确保高优先级消息能够更快地被消费的一种队列。它允许您为不同的消息设置优先级，并确保高优先级的消息能够尽快被消费。

使用场景：

处理紧急任务：当您有紧急任务需要处理时，可以使用优先级队列来确保这些任务能够被尽快消费和处理。
提高重要消息的处理效率：通过将重要消息标记为高优先级，您可以确保它们能够在其他消息之前进行处理。

优点：

灵活设置消息优先级：您可以根据消息的重要性和紧急性，为不同的消息设置合适的优先级。
提高重要消息的处理效率：高优先级的消息能够更快地被消费和处理，从而提高系统的响应速度和效率。

缺点：

优先级队列不能保证绝对的有序性，只能尽量按照优先级进行处理。
如果消息的优先级过多，可能会导致低优先级消息的处理延迟。

4. 镜像队列（Mirrored Queue）：

镜像队列是一种用于实现高可用性和故障转移的队列。它通过在多个节点上复制队列中的消息来确保即使某个节点宕机，消息也不会丢失。

使用场景：

提高可靠性：当您需要确保即使发生硬件故障或节点宕机，消息也不会丢失时，可以使用镜像队列来提高系统的可靠性。
实现故障转移：当某个节点发生故障时，镜像队列能够自动将消息路由到其他可用节点上，从而实现故障转移。

优点：

提供数据冗余和可靠性：通过在多个节点上复制队列中的消息，镜像队列能够提供数据冗余和容错能力，增强系统的稳定性。
自动故障转移：当发生节点故障时，镜像队列能够自动将消息路由到其他可用节点上，确保消息的持久性和可靠性。

缺点：

增加了网络开销和资源消耗：为了实现镜像队列，需要在多个节点之间复制消息，这可能会增加网络开销和系统资源消耗。

5. 批量队列（Batch Queue）：

批量队列是一种将多个小消息打包成一个大消息进行传输的队列，以减少网络开销和资源消耗。

使用场景：

提高传输效率：当您需要发送大量小消息时，可以使用批量队列将它们打包成一个大消息，从而减少传输开销。
节省系统资源：通过减少传输的次数和数据量，批量队列可以节省系统的网络带宽和处理资源。

优点：

提高传输效率：将多个小消息合并为一个大消息可以减少传输的次数和数据量，提高系统的传输效率。
节省系统资源：通过减少网络传输和处理的次数，批量队列可以节省系统的网络带宽和处理资源。

缺点：

需要额外使用插件或自定义的批量处理机制来实现批量队列。
可能会增加消息处理的复杂性，并且对于某些实时性要求较高的场景可能不适用。

RabbitMQ 队列和消息的 TTL

什么是队列和消息的 TTL

TTL（Time-To-Live）是指消息的生存时间，它决定了消息在队列中可以存在的最长时间。当消息的 TTL 过期后，RabbitMQ 将自动将其删除。通过设置队列的 TTL，可以控制队列中消息的最长存活时间。

除了设置队列的 TTL 外，RabbitMQ 还提供了为消息设置 TTL 的功能。这意味着可以为每条消息单独设置生存时间。当消息的 TTL 到期后，无论是否被消费，RabbitMQ 都会将其自动删除。

通过设置队列的 TTL 或消息的 TTL，可以管理消息的过期和清理。这对于缓存失效策略、数据清理和异步任务处理等场景非常有用。

需要注意的是，设置队列的 TTL 只对新进入队列的消息有效，已经存在于队列中的消息不受影响。而设置消息的 TTL 则对每条消息生效。

为队列设置 TTL

为队列设置 TTL 可以确保队列中的消息不会无限期地积压，从而避免性能问题。通过设置队列的 TTL，可以控制队列中消息的最长存活时间。

在 RabbitMQ 中，可以通过在创建队列时指定 x-message-ttl 参数来设置队列的 TTL。以下是一个示例：

Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 60000); // TTL 为 60 秒
channel.queueDeclare("my-queue", true, false, false, args);

上述代码将创建一个名为 my-queue 的队列，并将 TTL 设置为 60 秒。这意味着当消息在队列中存活超过 60 秒时，将被自动删除。

需要注意的是，设置队列的 TTL 只对新进入队列的消息有效。已经存在于队列中的消息不受影响。因此，如果要应用 TTL 对所有消息生效，可以使用消息的 TTL 功能。

消息的 TTL

除了为队列设置 TTL 外，RabbitMQ 还提供了为消息设置 TTL 的功能。通过设置消息的 TTL，可以对每条消息进行个别控制，确保消息在一定时间后被自动删除。

要为消息设置 TTL，可以在发送消息时设置 AMQP 属性 expiration 的值为 TTL 的毫秒数。以下是一个示例：

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .expiration("60000") // TTL 为 60 秒
        .build();
channel.basicPublish("", "my-queue", properties, message.getBytes());

上述代码将发送一条消息到名为 my-queue 的队列，并为该消息设置了 60 秒的 TTL。当消息在队列中存活超过 60 秒时，将被自动删除。

需要注意的是，如果同时设置了队列的 TTL 和消息的 TTL，消息的 TTL 不能大于队列的 TTL。如果消息的 TTL 大于队列的 TTL，那么消息的 TTL 将被忽略。

另外，需要注意的是，TTL 只能精确到毫秒级别，并且在设置 TTL 后，RabbitMQ 并不会立即将过期的消息删除，而是等到下一次扫描时才进行清理。