RabbitMq 消息队列笔记 > AMPQ 基础概念说明
延迟队列实现


1.使用死信队列

abbitmq 本身不支持延迟队列,但提供了实现延迟队列的必备条件。


原理

queue可以通过 x-message-ttl 参数设置过期时间,到了过期时间的消息就会被标记为 dead letter 状态。

过期的消息可以通过 x-dead-letter-exchange 和 x-dead-letter-routing-key 参数转发到另一个 exchange 中去。

注意:由于队列的先进先出特性,只有当过期的消息到了队列的顶端(队首),才会被真正的丢弃或者进入死信队列。所以在考虑使用RabbitMQ来实现延迟任务队列的时候,需要确保业务上每个任务的延迟时间是一致的。如果遇到不同的任务类型需要不同的延时的话,需要为每一种不同延迟时间的消息建立单独的消息队列。



2.使用延迟插件

(1)下载延迟插件

https://www.rabbitmq.com/community-plugins.html

注意:支持RabbitMQ 3.5.8及更高版本


(2)将解压后的ez结尾的文件放到/rabbitmq/plugins/目录下

(3)输入rabbitmq-plugins list查看延迟插件是否存在


(4)启用延迟插件

启动插件:rabbitmq-plugins enable rabbitmq_delayed_message_exchange

——(关闭插件:rabbitmq-plugins disable rabbitmq_delayed_message_exchange)


访问可视化界面,有如下则说明启动成功


(5)使用延时Exchange(消息发送到Exchange,Exchange等待指定时间后发送到匹配队列)

注意x-delayed-type参数必须有,不然会报错。

package com.dfyang.rabbitmq.dead;
import com.dfyang.rabbitmq.RabbitConnectionFactory;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.util.HashMap;
import java.util.Map;
public class DelayExchangeDemo {
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitConnectionFactory.getConnection();
        Channel channel = connection.createChannel();
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-delayed-type", "direct");
        channel.exchangeDeclare("delay.plugin.exchange", "x-delayed-message", true, false, arguments);
        channel.queueDeclare("delay.plugin.queue", true, false, false, null);
        channel.queueBind("delay.plugin.queue", "delay.plugin.exchange", "delay.plugin.routingKey");
        Map<String, Object> headers = new HashMap<>();
        //延迟10s后发送
        headers.put("x-delay", 10000); 
        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        builder.headers(headers);
        channel.basicPublish("delay.plugin.exchange", "delay.plugin.routingKey", builder.build(), "该消息将在10s后发送到队列".getBytes());
        channel.close();
        connection.close();
    }