RabbitMQ 延时队列插件安装与使用示例详解(基于 Delayed Message Plugin)

rabbitmq_delayed_message_exchange

wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v4.0.7/rabbitmq_delayed_message_exchange-v4.0.7.ez

docker cp /root/rabbitmq_delayed_message_exchange-v4.0.7.ez rabbitmq:/plugins

// 进入容器 我这里使用容器名字 也可以用容器id进入
docker exec -it rabbitmq /bin/bash
// 移动到plugins目录下
cd plugins
// 查看是否上传成功
ls

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

rabbitmq-plugins list

docker restart rabbitmq

docker exec -it <容器名称或ID> /bin/bash

[rabbitmq_management, rabbitmq_delayed_message_exchange].

docker restart <容器名称或ID>

docker exec <容器名称或ID> rabbitmq-plugins list

/**
     * 声明延迟队列（Queue 本身无 TTL，延迟由消息属性 x-delay 控制）
     * 注意：需要安装 RabbitMQ 的延迟消息插件（rabbitmq_delayed_message_exchange）
     */
    @Bean
    public Queue delayQueue() {
        return QueueBuilder.durable(getRetry().getQueue()).build();
    }
    /**
     * 声明延迟交换机：必须使用插件类型 x-delayed-message，并声明延迟基础类型为 direct
     */
    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(
                getRetry().getExchange(),      // 名称
                "x-delayed-message",           // 类型（延迟消息插件）
                true,                          // durable
                false,                         // auto-delete
                args                           // 参数：声明 delayed 类型为 direct
        );
    }
    /**
     * 绑定延迟队列到延迟交换机
     */
    @Bean
    public Binding delayBinding() {
        return BindingBuilder.bind(delayQueue())
                .to(delayExchange())
                .with(getRetry().getRoutingKey())
                .noargs();
    }

/**
     * 默认延迟 30 分钟（1800000 毫秒）
     */
    public void sendDelayedTask(String message) {
        this.sendDelayedTask(message, 30 * 60 * 1000);
    }
    /**
     * 发送延迟消息
     */
    public void sendDelayedTask(String message, long delayMillis) {
        MessageProperties properties = new MessageProperties();
        properties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
        properties.setHeader("x-delay", delayMillis); // 设置延迟时间（单位：毫秒）
        Message amqpMessage = new Message(message.getBytes(StandardCharsets.UTF_8), properties);
        rabbitTemplate.send(
                rabbitMqProperties.getConfig().getRetry().getExchange(),
                rabbitMqProperties.getConfig().getRetry().getRoutingKey(),
                amqpMessage
        );
        log.info("延迟任务已发送，延迟 {} s 后投递: {}", delayMillis/1000, message);
    }