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

🐕延迟队列

使用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生产-->张二发送成功");
    }

}

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

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

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

创建通用延时消息