赞
踩
死信队列:元素产生后没及时的被消费,一直存放在队列中。
延迟队列:延时队列,队列内部是有顺序的,其最重要的特性是延迟时间,是希望在指定时间到了以后或之前取出和处理,类似一个定时任务,但是比定时任务要节省资源。延时队列其实就是特殊的死信队列。
但是队列的特点是先进先出,如果对应不同的元素有不同的时间,那么如果存在延迟时间小的元素在延迟时间大的元素之后,那么就达不到对应的效果。
如何实现类似定时任务的效果呢?
就是当消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者才拿到这个消息进行消费。
1.延迟队列(TTL)+死信队列(DLX)
2.死信队列(DLX)+对应的插件
这个方法的目的是,先产生两个队列,其中一个队列作为死信队列,一个作为绑定死信的延迟队列。通过设置TTL(延迟时间:可以再这里设置,也可以在生产者针对消息进行设置,但是这种方法,要求一个死信队列存放的都是同一个延迟时间的消息)。
- /**
- * 构建TTL死信延迟的exchange,并对其进行命名
- */
- @Bean
- public DirectExchange delayDirectExchange() {
- return ExchangeBuilder
- .directExchange(QueueEnum.DELAY_QUEUE.getExchangeName())
- .durable(true)
- .build();
- }
-
- /**
- * 用来转发死信的exchange(也就是一个普通的exchange)
- */
- @Bean
- public DirectExchange delayBindExchange() {
- return ExchangeBuilder
- .directExchange(QueueEnum.DELAY_BIND_QUEUE.getExchangeName())
- .durable(true)
- .build();
- }
-
- /**
- * TTL死信延迟队列
- */
- @Bean
- public Queue delayQueue() {
- return QueueBuilder
- .durable(QueueEnum.DELAY_QUEUE.getQueueName())
- .withArgument("x-message-ttl", CommonConstant.OUT_ORDER_TIME)
- .withArgument("x-dead-letter-exchange", QueueEnum.DELAY_BIND_QUEUE.getExchangeName())
- .withArgument("x-dead-letter-routing-key", QueueEnum.DELAY_BIND_QUEUE.getPath())
- .build();
- }
-
- /**
- * 负责转发TTL的队列(也是一个普通的队列)
- */
- @Bean
- public Queue delayBindQueue() {
- return QueueBuilder
- .durable(QueueEnum.DELAY_BIND_QUEUE.getQueueName())
- .build();
- }
-
- /**
- * TTL死信转发绑定
- */
- @Bean
- Binding delayBindBinding (){
- return BindingBuilder
- .bind(delayBindQueue())
- .to(delayBindExchange())
- .with(QueueEnum.DELAY_BIND_QUEUE.getPath());
- }
-
- }
其中延迟TTL队列的重要的属性
x-message-ttl:延迟时间
x-dead-letter-exchange:绑定的死信交换机
x-dead-letter-routing-key: 绑定的路由,一般是对应死信队列名称。
其中生产者通过调用方法可以把消息的延迟时间加上去,mq取延迟时间最短的为准。
- /**
- * 发送延迟消息
- * @param msg 消息内容
- * @param delayTime 延迟时间,单位毫秒
- */
-
- private static AmqpTemplate amqpTemplate;
- public static void sendDelayMsg(String msg, Long delayTime,QueueEnum queueEnum){
- amqpTemplate.convertAndSend(queueEnum.getExchangeName(),queueEnum.getPath(),msg,message -> {
- message.getMessageProperties().setExpiration(delayTime.toString());
- return message;
- });
- }
这个是自己写的方法,其中生产者生产消息的最重要的方法是
convertAndSend()
上面演示了通过TTL+DLX实现的延迟队列,这样分的时间段越多,需要的交换机及队列也越多。前面的方式实现起来有点复杂,其实有更简单的实现方式,在RabbitMQ 3.5.7及以上的版本提供了一个插件(rabbitmq-delayed-message-exchange)来实现延迟队列功能。同时插件依赖Erlang/OPT 18.0及以上。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。