网站建设pqiw,江苏中淮建设集团有限公司网站,建设部标准网站,ip提取网站源码带后台RabbitMQ延时队列
什么是延时队列
延时队列顾名思义#xff0c;即放置在该队列里面的消息是不需要立即消费的#xff0c;而是等待一段时间之后取出消费。
应用场景
场景一#xff1a;在订单系统中#xff0c;一个用户下单之后通常有30分钟的时间进行支付#xff0c;如…RabbitMQ延时队列
什么是延时队列
延时队列顾名思义即放置在该队列里面的消息是不需要立即消费的而是等待一段时间之后取出消费。
应用场景
场景一在订单系统中一个用户下单之后通常有30分钟的时间进行支付如果30分钟之内没有支付成功那么这个订单将进行一场处理。这是就可以使用延时队列将订单信息发送到延时队列。
场景二用户希望通过手机远程遥控家里的智能设备在指定的时间进行工作。这时候就可以将用户指令发送到延时队列当指令设定的时间到了再将指令推送到智能设备。
实现方式
Rabbitmq实现延时队列一般而言有两种形式 第一种方式利用两个特性 Time To Live(TTL消息过期)、Dead Letter ExchangesDLX死信队列 第二种方式利用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 ExchangesDLX
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)并且 requeuefalse.
死信实战
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设置对列参数信息HashMapString, 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();}}
