RabbitMQ延时队列

RabbitMQ延时队列

什么是延时队列

延时队列顾名思义，即放置在该队列里面的消息是不需要立即消费的，而是等待一段时间之后取出消费。

应用场景

场景一：在订单系统中，一个用户下单之后通常有30分钟的时间进行支付，如果30分钟之内没有支付成功，那么这个订单将进行一场处理。这是就可以使用延时队列将订单信息发送到延时队列。

场景二：用户希望通过手机远程遥控家里的智能设备在指定的时间进行工作。这时候就可以将用户指令发送到延时队列，当指令设定的时间到了再将指令推送到智能设备。

实现方式

Rabbitmq实现延时队列一般而言有两种形式： 第一种方式：利用两个特性： Time To Live(TTL消息过期)、Dead Letter Exchanges（DLX死信队列） 第二种方式：利用rabbitmq中的插件x-delay-message

三、第一种：利用TTL DLX实现延时队列的方式 AMQP协议和RabbitMQ队列本身没有直接支持延迟队列功能，但是可以通过以下特性模拟出延迟队列的功能。

1、Time To Live(TTL)

RabbitMQ可以针对Queue设置x-expires 或者 针对Message设置 x-message-ttl，来控制消息的生存时间，如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)

A: 通过队列属性设置，队列中所有消息都有相同的过期时间。 B: 对消息进行单独设置，每条消息TTL可以不同。

2、Dead Letter Exchanges（DLX）

RabbitMQ的Queue可以配置x-dead-letter-exchange和x-dead-letter-routing-key（可选）两个参数，如果队列内出现了dead letter，则按照这两个参数重新路由转发到指定的队列。 x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange x-dead-letter-routing-key：出现dead letter之后将dead letter重新按照指定的routing-key发送

用一个具体案例来实现第一种方式：用户下订单后，如何在一分钟没有支付就取消订单

死信队列

死信的概念

死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer 将消息投递到 broker 或者直接到queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

应用场景:为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中.还有比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

死信的来源

• 消息 TTL 过期

• 队列达到最大长度(队列满了，无法再添加数据到 mq 中)

• 消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false.

死信实战

package com.zking.xiongda.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;import java.util.HashMap;
import java.util.Objects;@Configuration
public class QueueConfig {/*** 死信队列* @return*/@Beanpublic Queue deadQueue(){return new Queue("dead-queue");}/*** 死信交互机*/@Beanpublic Exchange deadExchange(){return new DirectExchange("dead-exchange");}/*** 绑定死信队列* @return*/@Beanpublic Binding bindingExchange(){return BindingBuilder.bind(deadQueue()).to(deadExchange()).with("dead-exchange.key").noargs();}/*** 普通队列* @return*/@Beanpublic Queue xioangdaQueue(){// hashmap设置对列参数信息HashMap<String, Object> hashMap = new HashMap<>();// 设置消息过期时间hashMap.put("x-message-ttl",20000);// 设置消息过期转换死信队列交换机hashMap.put("x-dead-letter-exchange","dead-exchange");// 设置改队列关联转移到交换机路由keyhashMap.put("x-dead-letter-routing-key","dead-exchange.key");// true是否持久，false：是否排他，false：是否自动删除return new Queue("xiongda-queue",true,false,false,hashMap);}/*** 普通交换机*/@Beanpublic Exchange xioangdaExchange(){return new DirectExchange("xiongda-exchange");}/*** 绑定普通队列* @return*/@Beanpublic Binding xioangdaDindingExchange(){return BindingBuilder.bind(xioangdaQueue()).to(xioangdaExchange()).with("xiongda-exchange.key").noargs();}}