目录

• 1、DelayExchange插件配置
• • 1.1、下载DelayExchange插件
  • 1.2、安装DelayExchange插件
  • 1.3、启动DelayExchange插件
• 2、使用原理
• 3、使用说明
• 4、完整代码
• 5、测试效果

1、DelayExchange插件配置

1.1、下载DelayExchange插件

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

选择对应版本进行下载，比如：rabbitmq_delayed_message_exchange-3.8.17.8f537ac.ez

1.2、安装DelayExchange插件

将下载的文件拷贝到服务器，比如root目录下；

docker环境操作：

docker搭建时，拷贝到容器的/opt/rabbitmq/plugins/中docker cp rabbitmq_delayed_message_exchange-3.9.0.ez rabbitmq:/opt/rabbitmq/plugins/

linux环境操作：

# 如果不是docker搭建，则拷贝到/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.27/plugins/目录下cp rabbitmq_delayed_message_exchange-3.9.0.ez /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.27/plugins

1.3、启动DelayExchange插件

docker环境操作：

docker exec -it rabbitmq bashrabbitmq-plugins enable rabbitmq_delayed_message_exchange

linux环境操作：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

2、使用原理

DelayExchange需要将一个交换机声明为delayed类型。当我们发送消息到delayExchange时，流程如下：

• 接收消息
• 判断消息是否具备x-delay属性
• 如果有x-delay属性，说明是延迟消息，持久化到硬盘，读取x-delay值，作为延迟时间
• 返回routing not found结果给消息发送者
• x-delay时间到期后，重新投递消息到指定队列

3、使用说明

声明一个交换机，交换机的类型可以是任意类型，通过@RabbitListener注解绑定交换机并且设定delayed属性为true即可，如下：

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "queue_delay"),     exchange = @Exchange(name = "exchange_delay", delayed = "true"), key = "routing_delay"))

发送消息时，通过MessageBuilder组织消息，并且设置设置header的key为x-delayvalue为TTL，TTL单位为毫秒，如下：

Message message = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8)).setHeader("x-delay", 1000).build();

4、完整代码

Controller：

    /**     * 测试延时消息     * 由于使用了延时队列，所有再回调方法中的setReturnCallback打印异常，可以通过if判断过虑延时队列使用的outingKey即可     */    @PostMapping("/sendDelay")    public String sendDelay() { String msg = "发送时间：" + LocalDateTime.now() + "，延时："; for (int i = 1; i <= 10; i++) {     String sendMsg = msg + i + "秒";     long time = i * 1000;     rabbitTemplate.convertAndSend(RabbitmqContext.EXCHANGE_DELAY, RabbitmqContext.ROUTING_DELAY, sendMsg, message -> {  //注意这里时间可以使long，而且是设置header  message.getMessageProperties().setHeader("x-delay", time);  return message;     }); } return "发送成功...";    }

Config：

@Configurationpublic class DelayConfig {    /**     * 配置队列名称     *     * @return     */    @Bean    public Queue delayQueue() { return new Queue(RabbitmqContext.QUEUE_DELAY);    }    /**     * 延时队列可以配置任意类型的交换机     *     * @return     */    @Bean    public CustomExchange delayExchange() { Map<String, Object> args = new HashMap<>(2); args.put("x-delayed-type", "direct"); // x-delayed-message 固定写法 return new CustomExchange(RabbitmqContext.EXCHANGE_DELAY, "x-delayed-message", true, false, args);    }    @Bean    public Binding delayBinding() { return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(RabbitmqContext.ROUTING_DELAY).noargs();    }}

Listener：

@Component@Slf4jpublic class DelayReceiveListener {    /**     * 设置延时队列     *     * @param msg     接收的文本消息     * @param channel 通道信息     * @param message 附加的参数信息     */    @RabbitListener(queues = "delay_queue")    public void receiveQueue(String msg, Channel channel, Message message) { System.out.println("这个是延时消息：" + msg + "，当前时间：" + LocalDateTime.now());    }}

5、测试效果