[MQ] 延迟队列/延迟插件下载

Posted 沫洺

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了[MQ] 延迟队列/延迟插件下载相关的知识,希望对你有一定的参考价值。

✨✨个人主页:沫洺的主页

📚📚系列专栏: 📖 JavaWeb专栏📖 JavaSE专栏 📖 Java基础专栏📖vue3专栏 

                           📖MyBatis专栏📖Spring专栏📖SpringMVC专栏📖SpringBoot专栏

                           📖Docker专栏📖Reids专栏📖MQ专栏📖SpringCloud专栏     

💖💖如果文章对你有所帮助请留下三连✨✨

🐕延迟队列

使用rabbitmq的延时队列插件,实现同一个队列中有多个不同超时时间的消息,并按时间超时顺序出队

🦢下载延迟插件

在 RabbitMQ 的 3.5.7 版本之后,提供了一个插件(rabbitmq-delayed-message-exchange)来实现延迟队列 ,同时需保证 Erlang/OPT 版本为 18.0 之后。

我这里 MQ 的版本是 3.10.5,现在去 GitHub 上根据版本号下载插件

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases 

 安装插件并启用

下载完成后直接把插件放在 /root/211 目录,然后拷贝到容器内plugins目录下(rabbitmq是容器的name,也可以使用容器id)

 

docker cp /home/211/rabbitmq_delayed_message_exchange-3.10.0.ez rabbitmq:/plugins

进入 Docker 容器

docker exec -it rabbitmq /bin/bash

在plugins内启用插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

退出容器

exit

重启 RabbitMQ

docker restart rabbitmq

安装成功

 通过UI查看

🐬SpringBoot使用延迟队列

消费者

自定义交换机CustomExchange

@Component
public class DelayConsumer 
    private static final String ENAME = "211-DelayExchage-01";
    private static final String QNAME1 = "211-DelayQueue-01";

    //自定义交换机
    @Bean
    public CustomExchange customExchange() 
        HashMap<String, Object> args = new HashMap<>();
        args.put("x-delayed-type","direct");
        //延迟交换机
        return new CustomExchange(ENAME, "x-delayed-message", true, false, args);
    

    //定义一个队列
    @Bean
    public Queue queue() 
        return QueueBuilder.durable(QNAME1).build();
    

    //创建队列和交换机的绑定关系
    @Bean
    public Binding binding1() 
        return BindingBuilder.bind(queue()).to(customExchange()).with("diancan").noargs();
    

    //消费者
    @RabbitHandler
    @RabbitListener(queues = QNAME1)
    public void process1(UserRegisterOk userRegisterOk) 
        System.out.println(DateUtil.format(DateUtil.date(),"HH:mm:ss") +"消费者收到:" + userRegisterOk.getName() + "," + userRegisterOk.getHeight());
    

messages delayed:0

默认延迟时间0s

生产者

设置延迟时间

        message -> 
            //设置消息延迟时间5秒,5秒之后投递给队列 针对的是交换机
            message.getMessageProperties().setDelay(5*1000);
            return message;
        
@Component
public class DelayProducer 

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage()
        //延迟5秒
        UserRegisterOk userRegisterOk1 = UserRegisterOk.builder().name("张一").phone("123456").height("1.8.5").build();
        //要将对象序列化,转成字符串,使用消息转换器MessageConverter
        rabbitTemplate.convertAndSend("211-DelayExchage-01","diancan",userRegisterOk1,message -> 
            message.getMessageProperties().setDelay(5*1000);
            return message;
        );
        System.out.println(DateUtil.format(DateUtil.date(),"HH:mm:ss") +"生产者1生产-->张一发送成功");
        
        //延迟8秒
        UserRegisterOk userRegisterOk2 = UserRegisterOk.builder().name("张二").phone("123456").height("1.8.5").build();
        //要将对象序列化,转成字符串,使用消息转换器MessageConverter
        rabbitTemplate.convertAndSend("211-DelayExchage-01","diancan",userRegisterOk2,message -> 
            message.getMessageProperties().setDelay(8*1000);
            return message;
        );
        System.out.println(DateUtil.format(DateUtil.date(),"HH:mm:ss") +"生产者2生产-->张二发送成功");
    

整个的流程就是生产者生产消息后,在交换机中停留指定的延迟时间,后发送到队列,消费者获取队列中的消息 

补充延迟队列不常用的两种方式

创建具有超时功能且绑定死信交换机的消息队列

创建通用延时消息

以上是关于[MQ] 延迟队列/延迟插件下载的主要内容,如果未能解决你的问题,请参考以下文章

MQ 死信队列/延迟队列-( 商品秒杀后30分钟之内付款)

MQ 死信队列/延迟队列-( 商品秒杀后30分钟之内付款)

RabbitMQ-消息可靠性&延迟消息

一篇带您搞懂MQ延迟队列实战操作

一篇带您搞懂MQ延迟队列实战操作

RabbitMQ延迟消息问题(含Demo工程)