当前位置：首页 > news >正文

RabbitMQ--延迟队列事务消息分发

news 来源：原创 2025/7/4 20:32:58

1.延迟队列

1.1应用场景

1.2利用TTL+死信队列模拟延迟队列存在的问题

1.3延迟队列插件

1.4常见面试题

2.事务

2.1配置事务管理器

3.消息分发

3.1概念

3.2应用场景

3.2.1限流

3.2.2负载均衡

1.延迟队列

延迟队列(Delayed Queue)，即消息被发送以后, 并不想让消费者立刻拿到消息, 而是等待特定时间后, 消费者才能拿到这个消息进行消费

1.1应用场景

延迟队列的应用场景有很多，比如：

. 智能家居: 用户希望通过手机远程遥控家⾥的智能设备在指定的时间进行⼯作. 这时候就可以将用户指令发送到延迟队列, 当指令设定的时间到了再将指令推送到智能设备.

日常管理: 预定会议后，需要在会议开始前⼗五分钟提醒参会人参加会议

用户注册成功后, 7天后发送短信, 提高用户活跃度等......

RabbitMQ本⾝没有直接支持延迟队列的的功能, 但是可以通过前面所介绍的TTL+死信队列的方式组合模拟出延迟队列的功能.

假设⼀个应用中需要将每条消息都设置为10秒的延迟, 生产者通过 normal_exchange 这个交换器将发送的消息存储在 normal_queue 这个队列中. 消费者订阅的并非是 normal_queue 这个队列, ⽽

是 dlx_queue 这个队列. 当消息从 normal_queue 这个队列中过期之后被存入 dlx_queue 这个

队列中，消费者就恰巧消费到了延迟10s的这条消息

1.2利用TTL+死信队列模拟延迟队列存在的问题

如果我们先发送20s过期的消息再发送10s过期的消息，我们会发现10s过期的消息，也是在20s后进入到了死信队列。

消息过期之后, 不⼀定会被马上丢弃. 因为RabbitMQ只会检查队首消息是否过期, 如果过期则丢到死信队列. 此时就会造成⼀个问题, 如果第⼀个消息的延时时间很长, 第⼆个消息的延时时间很短, 那第⼆个消息并不会优先得到执行.

所以在考虑使⽤TTL+死信队列实现延迟任务队列的时候, 需要确认业务上每个任务的延迟时间是⼀致的, 如果遇到不同的任务类型需要不同的延迟的话, 需要为每⼀种不同延迟时间的消息建⽴单独的消息队列

1.3延迟队列插件

RabbitMQ官⽅也提供了⼀个延迟的插件来实现延迟的功能

参考: https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq

1.4常见面试题

延迟队列作为RabbitMQ的⾼级特性,也是面试的⼀⼤重点.

介绍下RabbitMQ的延迟队列

延迟队列是⼀个特殊的队列, 消息发送之后, 并不立即给消费者, ⽽是等待特定的时间, 才发送给消费者.

延迟队列的应⽤场景有很多, 比如:

1. 订单在⼗分钟内未⽀付自动取消

2. 用户注册成功后, 3天后发调查问卷

3. 用户发起退款, 24小时后商家未处理, 则默认同意, 自动退款

4. ......

但RabbitMQ本⾝并没直接实现延迟队列, 通常有两种方法:

1. TTL+死信队列组合的方式

2. 使⽤官⽅提供的延迟插件实现延迟功能

⼆者对⽐:

1. 基于死信实现的延迟队列

a. 优点: 1) 灵活不需要额外的插件⽀持

b. 缺点: 1) 存在消息顺序问题 2) 需要额外的逻辑来处理死信队列的消息, 增加了系统的复杂性

2. 基于插件实现的延迟队列

a. 优点: 1) 通过插件可以直接创建延迟队列, 简化延迟消息的实现. 2) 避免了DLX的时序问题

b. 缺点: 1) 通过依赖特定的插件，有运维工作 2) 只适用于特定版本

2.事务

RabbitMQ是基于AMQP协议实现的, 该协议实现了事务机制, 因此RabbitMQ也⽀持事务机制. Spring AMQP也提供了对事务相关的操作. RabbitMQ事务允许开发者确保消息的发送和接收是原子性的, 要么全部成功, 要么全部失败.

2.1配置事务管理器

@Configuration
public class TransactionConfig {@Beanpublic RabbitTransactionManager 
transactionManager(CachingConnectionFactory connectionFactory){return new RabbitTransactionManager(connectionFactory);}@Beanpublic RabbitTemplate rabbitTemplate(CachingConnectionFactory 
connectionFactory){RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setChannelTransacted(true);return rabbitTemplate;}
}

3.消息分发

3.1概念

RabbitMQ队列拥有多个消费者时, 队列会把收到的消息分派给不同的消费者. 每条消息只会发送给订阅列表里的⼀个消费者. 这种方式非常适合扩展, 如果现在负载加重,那么只需要创建更多的消费者来消费处理消息即可.

默认情况下, RabbitMQ是以轮询的方法进行分发的, 而不管消费者是否已经消费并已经确认了消息. 这种方式是不太合理的, 试想⼀下, 如果某些消费者消费速度慢, 而某些消费者消费速度快, 就可能会导致某些消费者消息积压, 某些消费者空闲, 进⽽应用整体的吞吐量下降.

如何处理呢? 我们可以使⽤前⾯章节讲到的channel.basicQos(int prefetchCount) ⽅法, 来限制当前信道上的消费者所能保持的最⼤未确认消息的数量

⽐如: 消费端调⽤了 channelbasicQos(5) , RabbitMQ会为该消费者计数, 发送⼀条消息计数+1, 消费⼀条消息计数-1, 当达到了设定的上限, RabbitMQ就不会再向它发送消息了,直到消费者确认了某条消息.类似TCP/IP中的"滑动窗口".

prefetchCount 设置为0时表示没有上限.

basicQos 对拉模式的消费无效

3.2应用场景

1.限流

2.非公平分发

3.2.1限流

如下使用场景：

订单系统每秒最多处理5000请求, 正常情况下, 订单系统可以正常满足需求

但是在秒杀时间点, 请求瞬间增多, 每秒1万个请求, 如果这些请求全部通过MQ发送到订单系统, 无疑会把订单系统压垮.

RabbitMQ提供了限流机制, 可以控制消费端⼀次只拉取N个请求

通过设置prefetchCount参数, 同时也必须要设置消息应答方式为手动应答

prefetchCount: 控制消费者从队列中预取(prefetch)消息的数量, 以此来实现流控制和负载均衡

3.2.2负载均衡

我们也可以用此配置,来实现"负载均衡"

如下图所示, 在有两个消费者的情况下，⼀个消费者处理任务非常快, 另⼀个非常慢,就会造成⼀个消费者会⼀直很忙, 而另⼀个消费者很闲. 这是因为 RabbitMQ 只是在消息进⼊队列时分派消息. 它不考虑消费者未确认消息的数量

我们可以使⽤设置prefetch=1 的⽅式, 告诉 RabbitMQ ⼀次只给⼀个消费者⼀条消息, 也就是说, 在处理并确认前⼀条消息之前, 不要向该消费者发送新消息. 相反, 它会将它分派给下⼀个不忙的消费者

1.延迟队列

1.1应用场景

1.2利用TTL+死信队列模拟延迟队列存在的问题

1.3延迟队列插件

1.4常见面试题

2.事务

2.1配置事务管理器

3.消息分发

3.1概念

3.2应用场景

3.2.1限流

3.2.2负载均衡

相关文章：