或许是全网最全的延迟队列

2023-12-15 22:33:28

什么是延迟队列

作用：用来存储延迟消息
延迟消息：生产者发送一个消息给mq，然后mq会经过一段时间（延迟时间），然后在把这个消息发送给消费者

应用场景

预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议
推送某些数据的定时任务
微信公众号文章的延迟发布
订单超时未支付自动取消订单

实现延迟队列

在rabbitmq中没有提供真正意义上的延迟队列。要实现延迟队列有两套方案

方案一：基于死信队列中的消息TTL过期模式的进行改造，不监听对应队列，使消息过期后全部进入死信队列以达成延时效果，主要有队列TTL和消息TTL两种
方案二：使用延时队列插件，让交换机管理消息延时时间（常用）

创建工程

创建springBoot工程，勾选需要的依赖

添加RabbitMQ配置

spring.rabbitmq.host=xxxx
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=DeadQueue

使用TTL+死信队列

队列TTL案例

对队列QA设置过期时间 10S，队列QB设置过期时间 40S，不监听QA、QB队列，使消息进入队列后不被消费导致TTL超时进入QD延迟队列

Y是死信交换机，QD是死信队列

对队列设置TTL
缺点：每增加一个新的时间需求，就要新增一个队列
创建RabbitMQ配置文件

package com.dmbjz.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/* RabbitMQ的交换机、队列配置文件 */
@Configuration
public class ExchangeQueueConfig {

    public static final String X_EXCHANGE = "X";
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    public static final String DEAD_LETTER_QUEUE = "QD";

    /*创建X交换机*/
    @Bean
    public DirectExchange xExchange(){
        return new DirectExchange(X_EXCHANGE);
    }

    /*创建死信交换机*/
    @Bean
    public DirectExchange yExchange(){
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    //声明队列 A ttl 为 10s 并绑定到对应的死信交换机
    @Bean("queueA")
    public Queue queueA(){
        Map<String, Object> args = new HashMap<>(3);
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);      //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-routing-key", "YD");                     //声明当前队列的死信路由 key
        args.put("x-message-ttl", 10000);                                //声明队列的 TTL
        return QueueBuilder.durable(QUEUE_A).withArguments(args).build();
    }


    // 声明队列 A 绑定 X 交换机
    @Bean
    public Binding queueaBindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }


    //声明队列 B ttl 为 40s 并绑定到对应的死信交换机
    @Bean("queueB")
    public Queue queueB(){
        Map<String, Object> args = new HashMap<>(3);
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);         //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-routing-key", "YD");                        //声明当前队列的死信路由 key
        args.put("x-message-ttl", 40000);                                   //声明队列的 TTL
        return QueueBuilder.durable(QUEUE_B).withArguments(args).build();
    }

    //声明队列 B 绑定 X 交换机
    @Bean
    public Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queue1B).to(xExchange).with("XB");
    }

    //声明死信队列 QD
    @Bean("queueD")
    public Queue queueD(){
        return new Queue(DEAD_LETTER_QUEUE);
    }

    //声明死信队列 QD 绑定关系
    @Bean
    public Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,
                                        @Qualifier("yExchange") DirectExchange yExchange){
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }
}

生产者代码：

package com.dmbjz.controller;


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;
import java.util.Date;

/* 生产者发送消息Controller */
@RestController
@RequestMapping("/ttl")
@Slf4j
public class SendMessageController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RequestMapping("/sendMessage/{message}")
    public void sendMsg(@PathVariable String message){

        log.info("当前时间:{},发送一条信息给两个TTL队列,消息内容:{}",new Date(),message);
        rabbitTemplate.convertAndSend("X","XA",message.getBytes(StandardCharsets.UTF_8));
        rabbitTemplate.convertAndSend("X","XB",message.getBytes(StandardCharsets.UTF_8));

    }
}

消费者代码：

package com.dmbjz.consumer;


import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/* 队列TTL消费者 */
@Component
@Slf4j
public class DeadLetterQueueConsumer {


    @RabbitListener(queues = "QD")
    public void receiveD(Message message, Channel channel)throws Exception{
        String msg = new String(message.getBody());
        log.info("当前时间:{},收到死信队列的消息:{}",new Date(),msg);
    }
}

浏览器访问地址测试：

http://localhost:8080/ttl/sendMessage/测试消息TTL

消息TTL案例

对消息设置过期时间，不监听QC队列，消息超时后自动进入QD延迟队列
缺点：如果积压在队列前面的消息延时时长很长，而后面积压的消息延时时长很短，积压时间短的消息并不会被提前放入死信队列；如果QC恰好又设置了积压上限，无法被积压的消息将直接进入延时队列，达不到延时效果
对消息设置TTL
修改配置文件：

    //声明队列 QC
    @Bean
    public Queue queueC(){
        Map<String, Object> args = new HashMap<>(3);
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);      //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-routing-key", "YD");                     //声明当前队列的死信路由 key
        return QueueBuilder.durable(QUEUE_C).withArguments(args).build();
    }

    //声明队列 QC 绑定 X 交换机
    @Bean
    public Binding queuebCBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueC()).to(xExchange).with("XC");
    }

生产者代码：

    //声明队列 QC
    @Bean
    public Queue queueC(){
        Map<String, Object> args = new HashMap<>(3);
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);      //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-routing-key", "YD");                     //声明当前队列的死信路由 key
        return QueueBuilder.durable(QUEUE_C).withArguments(args).build();
    }

    //声明队列 QC 绑定 X 交换机
    @Bean
    public Binding queuebCBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueC()).to(xExchange).with("XC");
    }

浏览器访问地址进行测试：

http://localhost:8080/ttl/sendMessagExpira/测试消息1/10000
http://localhost:8080/ttl/sendMessagExpira/测试消息2/1000

延时插件

使用延时队列插件实现延时队列功能，原理为交换机管理消息延时时间
插件版本需要兼容 RabbitMQ 版本，具体参考其发布说明**，**延时队列插件下载：github
插件安装步骤

1.将安装目录的延时队列插件拷贝到RabbitMQ插件目录
	cp rabbitmq_delayed_message_exchange-3.8.0.ez /root/rabbitmq_server-3.8.8/plugins
    
    
2.安装延时队列插件   
	rabbitmq-plugins enable rabbitmq_delayed_message_exchange
    

3、重启RabbitMQ服务
	systemctl restart rabbitmq-server

延时队列插件安装完成
重启服务后交换机多了延迟类型
案例演示：
延时队列插件实际落地固定为图中架构模式
延时队列插件架构图
创建配置文件：

package com.dmbjz.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/* 延时队列插件案例 RabbitMQ配置类 */
@Configuration
public class DelayedQueueConfig {


    private static final String delayed_queue_name = "delayed.queue";
    private static final String delayed_exchange_name = "delayed.exchange";
    private static final String delayed_routingkey = "delayed.routingkey";


    /*创建延时插件的交换机，需要使用自定义方法进行创建
    *   插件版非死信队列，不需要路由到不同的交换机进行指定过期时间，所以固定为 direct 类型交换机
    * */
    @Bean
    public CustomExchange delayedExchange(){

        Map<String,Object> map = new HashMap<>(1);
        map.put("x-delayed-type","direct");       //延迟队列类型，固定值

        return new CustomExchange(delayed_exchange_name,"x-delayed-message",
                true,false,map);

    }

    /*队列*/
    @Bean
    public Queue delayQueue(){
        return QueueBuilder.durable(delayed_queue_name).build();
    }

    /*绑定，自定义交换机绑定多一个 noargs方法 */
    @Bean
    public Binding delayBing(@Qualifier("delayQueue") Queue delayQueue,
                             @Qualifier("delayedExchange") CustomExchange delayedExchange){
        return BindingBuilder.bind(delayQueue).to(delayedExchange)
                .with(delayed_routingkey)
                .noargs();
    }
}

生产者代码：

    /*延时插件案例*/
    @RequestMapping("/sendMessagPlugin/{message}/{time}")
    public void sendMsgPlugin(@PathVariable String message,
                              @PathVariable Integer time){

        MessageProperties properties = new MessageProperties();
        properties.setDelay(time);      //设置延时时间
        Message msg = new Message(message.getBytes(StandardCharsets.UTF_8),properties);

        log.info("当前时间:{},发送具有过期时间为{}毫秒的信息给延时插件队列,消息内容:{}",new Date(),time,message);
        rabbitTemplate.convertAndSend("delayed.exchange","delayed.routingkey",msg);

    }

消费者代码：

package com.dmbjz.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/* 延时队列插件 消费者 */
@Component
@Slf4j
public class DelayQueueConsumer {

    @RabbitListener(queues = "delayed.queue")
    public void receiveDelayQueue(Message message, Channel channel)throws Exception{

        String msg = new String(message.getBody());
        log.info("当前时间:{},收到死信队列的消息:{}",new Date(),msg);

    }
}

浏览器访问地址进行测试：

http://localhost:8080/ttl/sendMessagPlugin/测试消息1/10000
http://localhost:8080/ttl/sendMessagPlugin/测试消息2/1000

文章来源:https://blog.csdn.net/weixin_61427900/article/details/135025599
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！