Rabbitmq 通过延迟插件实现延迟队列

# 开启插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

# 重启 rabbitmq
/sbin/service rabbitmq-server restart

# 查看插件是否安装成功
sudo rabbitmq-plugins list

@Configuration
public class RabbitConfig implements ApplicationContextAware {
    
    private ApplicationContext applicationContext;    
    
    @PostConstruct
    public RabbitAdmin rabbitAdmin() {
        RabbitAdmin rabbitAdmin = applicationContext.getBean("rabbitAdmin", RabbitAdmin.class);
        TopicExchange exchange = new TopicExchange("exchange.delay");
        // 交换器设置延迟属性
        exchange.setDelayed(true);				                  
        rabbitAdmin.declareQueue(new Queue("queue.delay"));
        rabbitAdmin.declareExchange(exchange);
        rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue("queue.delay")).to(exchange).with("rountingkey.delay");
        return rabbitAdmin;
    }
}

 // 消息发送器                                  
public class SendMQService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    // 发送消息
    public void sendMessage(String exchange, String routingKey, String msg, Integer expirationTime) {
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, message -> {
            // 给每条消息设置过期时间
            message.getMessageProperties().setExpiration(expirationTime);
            return message;
        });
    }
}

// 消息监听器, 交换器 delayed = "true"
@Component
@RabbitListener(containerFactory = "listenerContainerFactory",
    bindings = @QueueBinding(value = @Queue(value = "queue.delay"),
        exchange = @Exchange(value = "exchange.delay", type = ExchangeTypes.TOPIC, delayed = "true"),
        key = "rountingkey.delay"))
@Slf4j
public class MsgListener {

    @RabbitHandler
    public void msgHandler(String msg) {
        log.info("接收到的延迟消息 [{}]",msg)
    }
}