rabbitmq延迟队列的使用方式

 @Bean
    public Queue directQueueLong(){
        return   QueueBuilder.durable("业务队列名称")
                .deadLetterExchange("死信交换机名称")
                .deadLetterRoutingKey("死信队列 RoutingKey")
                .ttl(20000) // 消息停留时间
                //.maxLength(500)
                .build();
    }

rabbitTemplate.convertAndSend("交换机名称", "RoutingKey","对象",
	message => {
		message.getMessageProperties().setExpiration(String.valueOf(5000))
			        return message;
	            }
           );

docker cp /opt/rabbitmq/rabbitmq_delayed_message_exchange-3.10.0.ez rabbitmq:/plugins

docker exec -it rabbitmq /bin/bash

#先执行，解除防火墙限制，增加文件权限
cd plugins
umask 0022
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

exit

docker restart rabbitmq

/*
 * Copyright (c) 2020, 2024, fpl1116.cn All rights reserved.
 *
 */
package com.fpl.consumers;
import cn.hutool.core.map.MapUtil;
import com.fpl.model.OrderingOk;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Map;
/**
 * <p>Project: spring-rabbitmq - DelayConsumer</p>
 * <p>Powered by fpl1116 On 2024-04-09 11:34:07</p>
 * <p>描述：<p>
 *
 * @author penglei
 * @version 1.0
 * @since 1.8
 */
@Configuration
@Slf4j
public class DelayConsumer {
    @Bean
    public Queue delayQueue1(){
        return   QueueBuilder.durable("Delay_Q01").lazy().build();
    }
    @Bean
    public CustomExchange delayExchange(){
        //参数x-delayed-type
        Map<String, Object> map = MapUtil.of("x-delayed-type","direct");
        return new CustomExchange("Delay_E01","x-delayed-message",true,false,map);
    }
    @Bean
    public Binding binding1(Queue delayQueue1, CustomExchange delayExchange){
        return BindingBuilder.bind(delayQueue1).to(delayExchange).with("RK01").noargs();
    }
//     @RabbitListener(queues = "Delay_Q01")
    public void receiveMessage(OrderingOk msg) {
        log.info("消费者1 收到消息："+ msg );
    }
}

/*
 * Copyright (c) 2020, 2024, fpl1116.cn All rights reserved.
 *
 */
package com.fpl.provider;
import com.fpl.model.OrderingOk;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
 * <p>Project: spring-rabbitmq - DelayProvider</p>
 * <p>Powered by fpl1116 On 2024-04-09 11:35:51</p>
 * <p>描述：<p>
 *
 * @author penglei
 * @version 1.0
 * @since 1.8
 */
@Service
public class DelayProvider {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void send(OrderingOk orderingOk) {
        rabbitTemplate.convertAndSend("Delay_E01", "RK01", orderingOk,new MessagePostProcessor(){
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                int id  = orderingOk.getId();
                int ttl = 0;
                if(id == 1){
                    ttl = 50*1000;
                }else if(id == 2){
                    ttl = 30*1000;
                }else if(id ==3){
                    ttl = 40*1000;
                }else if(id ==4){
                    ttl = 10*1000;
                }else if(id ==5){
                    ttl = 20*1000;
                }
                //延迟交换机使用的delay参数,设置消息的延期时长，单位是微妙
                message.getMessageProperties().setDelay(ttl);
                //延迟交换机消息默认是持久化的
                message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
                return message;
            }
        });
    }
}

@Test
    void test5() throws IOException {
        for (int i = 1; i <=5;i++){
            OrderingOk orderingOk = OrderingOk.builder().id(i).name("张 " + i).build();
            delayProvider.send(orderingOk);
            System.out.println("发送成功:"+i);
        }
        System.in.read();
    }