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

RabbitMQ延时队列

article 2025/3/7 4:53:26

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
     */
    @Bean
    public Queue deadQueue(){
        return new Queue("dead-queue");
    }

    /**
     * 死信交互机
     */
    @Bean
    public Exchange deadExchange(){
        return new DirectExchange("dead-exchange");
    }

    /**
     * 绑定死信队列
     * @return
     */
    @Bean
    public Binding bindingExchange(){
        return BindingBuilder.bind(deadQueue())
                .to(deadExchange())
                .with("dead-exchange.key")
                .noargs();
    }


    /**
     * 普通队列
     * @return
     */
    @Bean
    public Queue xioangdaQueue(){
        // hashmap设置对列参数信息
        HashMap<String, Object> hashMap = new HashMap<>();
        // 设置消息过期时间
        hashMap.put("x-message-ttl",20000);
        // 设置消息过期转换死信队列交换机
        hashMap.put("x-dead-letter-exchange","dead-exchange");
        // 设置改队列关联转移到交换机路由key
        hashMap.put("x-dead-letter-routing-key","dead-exchange.key");

        // true是否持久，false：是否排他，false：是否自动删除
        return new Queue("xiongda-queue",true,false,false,hashMap);
    }

    /**
     * 普通交换机
     */
    @Bean
    public Exchange xioangdaExchange(){
        return new DirectExchange("xiongda-exchange");
    }

    /**
     * 绑定普通队列
     * @return
     */
    @Bean
    public Binding xioangdaDindingExchange(){
        return BindingBuilder.bind(xioangdaQueue())
                .to(xioangdaExchange())
                .with("xiongda-exchange.key")
                .noargs();
    }




}

查看全文

http://www.kler.cn/a/427666.html