【RabbitMQ】RabbitMQ高级：死信队列和延迟队列

admin • • 5G

目录

设置TTL（过期时间）

概述

在电商平台下单，订单创建成功，等待支付，一般会给30分钟的时间，开始倒计时。如果在这段时间内用户没有支付，则默认订单取消。

该如何实现？

  1. 定期轮询（数据库等）

用户下单成功，将订单信息放入数据库，同时将支付状态放入数据库，用户付款更改数据库状态。定期轮询数据库支付状态，如果超过30分钟就将该订单取消。

优点：设计实现简单

缺点：需要对数据库进行大量的IO操作，效率低下。

  1. Timer

Timer可以用来设置指定时间后执行的任务。

SimpleDateFormat simpleDateFormat=new SimpleDateFormat("HH:mm:ss");
Timer timer=new Timer();
TimerTask timerTask=new TimerTask(){
    @Override public void run(){
        System.out.println("用户没有付款，交易取消："+simpleDateFormat.format(new Date(System.currentTimeMillis())));
        timer.cancel();
    }
};
System.out.println("等待用户付款："+simpleDateFormat.format(new Date(System.currentTimeMillis())));
// 10秒后执行timerTask 
timer.schedule(timerTask, 10 * 1000);

缺点：

Timers没有持久化机制；不灵活 (只可以设置开始时间和重复间隔，对等待支付貌似够用)；不能利用线程池，一个timer一个线程；没有真正的管理计划。

  1. ScheduledExecutorService
SimpleDateFormat format=new SimpleDateFormat("HH:mm:ss"); 
// 线程工厂
ThreadFactory factory = Executors.defaultThreadFactory(); 
// 使用线程池 
ScheduledExecutorService service = new ScheduledThreadPoolExecutor(10, factory);
System.out.println("开始等待用户付款10秒：" + format.format(new Date()));
service.schedule(new Runnable() { 
    @Override public void run() { 
        System.out.println("用户未付款，交易取消：" + format.format(new Date())); 
    }
    // 等待10s 单位秒 
}, 10, TimeUnit.SECONDS);

优点：可以多线程执行，一定程度上避免任务间互相影响，单个任务异常不影响其它任务。

在高并发的情况下，不建议使用定时任务去做，因为太浪费服务器性能，不建议。

  1. RabbitMQ的TTL

  2. Quartz

  3. JCronTab

等等。

下面就重点介绍下，如何使用RabbitMQ的TTL

RabbitMQ使用TTL

TTL，Time to Live 的简称，即过期时间。

RabbitMQ 可以对消息和队列两个维度来设置TTL。

任何消息中间件的容量和堆积能力都是有限的，如果有一些消息总是不被消费掉，那么需要有一种过期的机制来做兜底。

目前有两种方法可以设置消息的TTL：

  1. 通过Queue属性设置，队列中所有消息都有相同的过期时间。
  2. 对消息自身进行单独设置，每条消息的TTL 可以不同。

如果两种方法一起使用，则消息的TTL 以两者之间较小数值为准。通常来讲，消息在队列中的生存时间一旦超过设置的TTL 值时，就会变成“死信”(Dead Message)，消费者默认就无法再收到该消息。当然，“死信”也是可以被取出来消费的，下一小节我们会讲解。

原生API案例

try{
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel())
        // 创建队列（实际上使用的是AMQP default这个direct类型的交换器） 
        // 设置队列属性 
        Map<String, Object> arguments=new HashMap<>();
        // 设置队列的TTL 
        arguments.put("x-message-ttl",30000);
        // 设置队列的空闲存活时间（如该队列根本没有消费者，一直没有使用，队列可以存活多久） 
        arguments.put("x-expires",10000);
        channel.queueDeclare(QUEUE_NAME,false,false,false,arguments);
        for(int i=0;i< 1000000;i++){
            String message="Hello World!"+i;
            channel.basicPublish("",QUEUE_NAME,new AMQP.BasicProperties().builder().expiration("30000").build(),message.getBytes());
            System.out.println(" [X] Sent '"+message+"'");
        }
}catch(TimeoutException e){
    e.printStackTrace();
}catch(IOException e){
    e.printStackTrace();
}

此外，还可以通过命令行方式设置全局TTL，执行如下命令:

rabbitmqctl set_policy TTL ".*" '{"message-ttl":30000}' --apply-to queues

还可以通过restful api方式设置，这里不做过多介绍。

默认规则：

  1. 如果不设置TTL，则表示此消息不会过期；
  2. 如果TTL设置为0，则表示除非此时可以直接将消息投递到消费者，否则该消息会被立即丢弃；

注意理解 message-ttl 、 x-expires 这两个参数的区别，有不同的含义。但是这两个参数属性都遵循上面的默认规则。一般TTL相关的参数单位都是毫秒（ms）。

springboot案例

在配置类里声明队列的时候设置TTL:

@Bean 
public Queue queueTTLWaiting() { 
    Map<String, Object> props = new HashMap<>(); 
    // 对于该队列中的消息，设置都等待10s 
    props.put("x-message-ttl", 10000); 
    Queue queue = new Queue("q.pay.ttl-waiting", false, false, false, props); 
    return queue; 
}

在生产者发消息时，可以指定消息的TTL：

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.UnsupportedEncodingException;

@RestController
public class PayController {
    @Autowired
    private AmqpTemplate rabbitTemplate;

    @RequestMapping("/pay/queuettl")
    public String sendMessage() {
        rabbitTemplate.convertAndSend("ex.pay.ttl-waiting", "pay.ttl-waiting", "发送了TTL-WAITING-MESSAGE");
        return "queue-ttl-ok";
    }

    @RequestMapping("/pay/msgttl")
    public String sendTTLMessage() throws UnsupportedEncodingException {
        MessageProperties properties = new MessageProperties();
        properties.setExpiration("5000");  // 设置消息的过期时间
        Message message = new Message("发送了WAITING- MESSAGE".getBytes("utf-8"), properties);
        rabbitTemplate.convertAndSend("ex.pay.waiting", "pay.waiting", message);
        return "msg-ttl-ok";
    }
}

死信队列

概述

死信队列，英文缩写是：DLX（Dead Letter Exchange），其实应该称为死信交换机更为合适。

当消息成为死信后，可以被重新发送到另一个交换机，这个交换机就是死信交换机。

在这里插入图片描述

实际上，死信队列就是普通的交换机，只不过我们人为的给其赋予了特殊的含义：当消息成为死信后，会重新发送到 DLX（死信交换机）。

默认情况下，当消息成为死信（过期、队列满了、消息 TTL 过期）的时候，RabbitMQ 会将这些消息进行清理，但是当配置了死信队列之后，RabbitMQ 会将死信发送到 DLX （死信交换机）中，这样就可以避免消息丢失。

死信队列的应用场景：

    • 为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入到死信队列中。
    • 用户在商城下单成功并进行支付活动，如果在指定的时候没有支付，将会将订单自动失效。

以下几种情况导致消息变为死信：

  1. 消息被拒绝（Basic.Reject/Basic.Nack），并且设置requeue参数为false；
  2. 消息过期；
  3. 队列达到最大长度。

对于RabbitMQ 来说，DLX 是一个非常有用的特性。它可以处理异常情况下，消息不能够被消费者正确消费(消费者调用了Basic.Nack 或者Basic.Reject)而被置入死信队列中的情况，后续分析程序可以通过消费这个死信队列中的内容来分析当时所遇到的异常情况，进而可以改善和优化系统。

原生API案例

try{
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        // 定义一个死信交换器（也是一个普通的交换器）

        channel.exchangeDeclare("exchange.dlx","direct",true);
        // 定义一个正常业务的交换器
        channel.exchangeDeclare("exchange.biz", "fanout", true);
        Map<String, Object> arguments = new HashMap<>();
        // 设置队列TTL
        arguments.put("x-message-ttl", 10000);
        // 设置该队列所关联的死信交换器（当队列消息TTL到期后依然没有消费，则加入死信队列）
        arguments.put("x-dead-letter-exchange", "exchange.dlx");
        // 设置该队列所关联的死信交换器的routingKey，如果没有特殊指定，使用原队列的 routingKey
        arguments.put("x-dead-letter-routing-key", "routing.key.dlx.test");
        channel.queueDeclare("queue.biz", true, false, false, arguments);
        channel.queueBind("queue.biz", "exchange.biz", "");
        channel.queueDeclare("queue.dlx", true, false, false, null);
        // 死信队列和死信交换器
        channel.queueBind("queue.dlx", "exchange.dlx", "routing.key.dlx.test");
        channel.basicPublish("exchange.biz", "", MessageProperties.PERSISTENT_TEXT_PLAIN, "dlx.test".getBytes());
} catch (Exception e) {
    e.printStackTrace();
}

springboot案例

下面通过设置TTL模拟在SpringBoot中如何使用死信队列。

修改RabbitConfig配置类，设置普通队列的属性（声明其死信队列和交换器），声明死信交换器，代码如下：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitConfig {

    @Bean
    public Queue queue() {
        Map<String, Object> props = new HashMap<>();
        // 消息的生存时间 10s
        props.put("x-message-ttl", 10000);
        // 设置该队列所关联的死信交换器（当队列消息TTL到期后依然没有消费，则加 入死信队列）
        props.put("x-dead-letter-exchange", "ex.go.dlx");
        // 设置该队列所关联的死信交换器的routingKey，如果没有特殊指定，使用原 队列的routingKey
        props.put("x-dead-letter-routing-key", "go.dlx");
        Queue queue = new Queue("q.go", true, false, false, props);
        return queue;
    }

    @Bean
    public Queue queueDlx() {
        Queue queue = new Queue("q.go.dlx", true, false, false);
        return queue;
    }

    @Bean
    public Exchange exchange() {
        DirectExchange exchange = new DirectExchange("ex.go", true, false, null);
        return exchange;
    }

    /**
     * 死信交换器
     *
     * @return
     */
    @Bean
    public Exchange exchangeDlx() {
        DirectExchange exchange = new DirectExchange("ex.go.dlx", true, false, null);
        return exchange;
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with("go").noargs();
    }

    /**
     * 死信交换器绑定死信队列
     * @return
     */
    @Bean
    public Binding bindingDlx() {
        return BindingBuilder.bind(queueDlx()).to(exchangeDlx()).with("go.dlx").noargs();
    }
}

在生产者端代码不用变。

如果想演示超过最大最列长度，可以设置普通对列长度：

Map<String, Object> props = MapUtil.newHashMap();
// 设置队列的最大长度
props.put("x-max-length", 10);

延迟队列

概述

延迟消息是指的消息发送出去后并不想立即就被消费，而是需要等（指定的）一段时间后才触发消费。

例如下面的业务场景：在支付宝上面买电影票，锁定了一个座位后系统默认会帮你保留15分钟时间，如果15分钟后还没付款那么不好意思系统会自动把座位释放掉。怎么实现类似的功能呢？

  1. 可以用定时任务每分钟扫一次，发现有占座超过15分钟还没付款的就释放掉。但是这样做很低效，很多时候做的都是些无用功；

  2. 可以用分布式锁、分布式缓存的被动过期时间，15分钟过期后锁也释放了，缓存key也不存在了；

  3. 还可以用延迟队列，锁座成功后会发送1条延迟消息，这条消息15分钟后才会被消费，消费的过程就是检查这个座位是否已经是“已付款”状态;

延迟队列的应用场景：

    • ① 订单在 10 分钟之内没有付款就自动取消。
    • ② 新创建的店铺，如果在 10 天之内都没有上传过商品，则自动发送消息提醒。
    • ③ 用户注册成功后，如果三天没有登录，则发送短信进行提醒。
    • ④ 用户发起退款，如果三天之内没有得到处理，则通知相关运营人员。
    • ⑤ 预定会议后，需要在预定的时间点前 10 分钟通知各个与会人员参加会议。

遗憾的是，在AMQP协议和RabbitMQ中都没有相关的规定和实现。

不过可以使用rabbitmq_delayed_message_exchange插件实现。

还可以我们可以借助上一小节介绍的“死信队列”来变相的实现。

插件和TTL方式有个很大的不同就是TTL存放消息在死信队列(delayqueue)里，二基于插件存放消息在延时交换机里(x-delayed-message exchange)。

插件实现延迟队列

安装插件

官网，下载 rabbitmq_delayed_message_exchange 插件，并解压到 RabbitMQ 的插件目录。

进入 RabbitMQ 的插件目录：

cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins

启用插件:

rabbitmq-plugins list 
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

重启rabbitmq-server:

systemctl restart rabbitmq-server

添加延迟队列插件之后：

在这里插入图片描述

代码

实现流程如下：

在这里插入图片描述

配置类RabbitmqConfig.java：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * 配置类，用来声明交换机和队列，并配置之间的关系
 */
@Configuration
public class RabbitmqConfig {

    /**
     * 普通交换机
     */
    public static final String EXCHANGE = "delayed.exchange";

    /**
     * routingkey
     */
    public static final String ROUTING_KEY = "delayed.routingkey";

    /**
     * 普通队列
     */
    public static final String QUEUE = "delayed.queue";

    @Bean
    public CustomExchange exchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(EXCHANGE, "x-delayed-message", true, false, args);
    }

    /**
     * 声明队列
     */
    @Bean
    public Queue queue() {
        return QueueBuilder.durable(QUEUE).build();
    }

    /**
     * 绑定关系
     */
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with(ROUTING_KEY).noargs();
    }
}

生产者ProducerController.java:

import com.github.config.RabbitmqConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;

/**
 * 生产者
 */
@Slf4j
@RestController
public class ProducerController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/send/{msg}/{ttl}")
    public String msg(@PathVariable("msg") String msg, @PathVariable("ttl") Integer ttl) {
        log.info("当前时间：{},发送一条时长{}毫秒 TTL 信息给队列:{}", LocalDateTime.now(), ttl, msg);
        MessagePostProcessor messagePostProcessor = (message) -> {
            // 注意，这里不再是 setExpiration ，而是 setDelay
            message.getMessageProperties().setDelay(ttl);
            return message;
        };
        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE, RabbitmqConfig.ROUTING_KEY, msg, messagePostProcessor);
        return "发送消息成功";
    }
}

消费者RabbitmqListener.java：

import com.github.config.RabbitmqConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;

/**
 * 消费者
 */
@Slf4j
@Component
public class RabbitmqListener {

    @RabbitListener(queues = RabbitmqConfig.QUEUE)
    public void receive(Message message) {
        log.info("当前时间：{},收到死信队列信息：{}", LocalDateTime.now(), new String(message.getBody(), StandardCharsets.UTF_8));
    }

}

演示：

curl 'http://127.0.0.1:8080/send/消息1/20000' -X GET
curl 'http://127.0.0.1:8080/send/消息2/2000' -X GET

IDEA 控制台结果显示：

在这里插入图片描述

TTL实现延迟队列

实现

实现过程如下：

在这里插入图片描述

配置类RabbitmqConfig.java：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 配置类，用来声明交换机和队列，并配置之间的关系
 */
@Configuration
public class RabbitmqConfig {

    /**
     * 普通交换机 X
     */
    public static final String EXCHANGE_X = "X";

    /**
     * 普通队列 QA
     */
    public static final String QUEUE_A = "QA";

    /**
     * 普通 routing key
     */
    public static final String ROUTING_KEY_XA = "XA";

    /**
     * 普通队列 QB
     */
    public static final String QUEUE_B = "QB";

    /**
     * 普通 routing key
     */
    public static final String ROUTING_KEY_XB = "XB";

    /**
     * 死信交换机 Y
     */
    public static final String DEAD_EXCHANGE_Y = "Y";

    /**
     * 死信队列 QD
     */
    public static final String DEAD_QUEUE_D = "QD";

    /**
     * 死信 routing key
     */
    public static final String DEAD_ROUTING_KEY_YD = "YD";

    /**
     * 声明交换机
     */
    @Bean
    public DirectExchange xExchange() {
        return new DirectExchange(EXCHANGE_X);
    }

    /**
     * 声明死信交换机
     */
    @Bean
    public DirectExchange yExchange() {
        return new DirectExchange(DEAD_EXCHANGE_Y);
    }

    /**
     * 声明队列
     */
    @Bean
    public Queue aQueue() {
        return QueueBuilder.durable(QUEUE_A)
                // 声明当前队列绑定的死信交换机
                .deadLetterExchange(DEAD_EXCHANGE_Y)
                // 声明当前队列绑定的死信队列
                .deadLetterRoutingKey(DEAD_ROUTING_KEY_YD)
                // 设置 TTL 时间
                .ttl(10 * 1000)
                .build();
    }

    /**
     * 声明队列
     */
    @Bean
    public Queue bQueue() {
        return QueueBuilder.durable(QUEUE_B)
                // 声明当前队列绑定的死信交换机
                .deadLetterExchange(DEAD_EXCHANGE_Y)
                // 声明当前队列绑定的死信队列
                .deadLetterRoutingKey(DEAD_ROUTING_KEY_YD)
                // 设置 TTL 时间
                .ttl(40 * 1000)
                .build();
    }

    /**
     * 声明死信队列
     */
    @Bean
    public Queue dQueue() {
        return QueueBuilder.durable(DEAD_QUEUE_D).build();
    }

    /**
     * 绑定关系
     */
    @Bean
    public Binding xaBinding() {
        return BindingBuilder.bind(aQueue()).to(xExchange()).with(ROUTING_KEY_XA);
    }

    /**
     * 绑定关系
     */
    @Bean
    public Binding xbBinding() {
        return BindingBuilder.bind(bQueue()).to(xExchange()).with(ROUTING_KEY_XB);
    }

    /**
     * 绑定关系
     */
    @Bean
    public Binding ydBinding() {
        return BindingBuilder.bind(dQueue()).to(yExchange()).with(DEAD_ROUTING_KEY_YD);
    }

}

生产者ProducerController.java：

import com.github.config.RabbitmqConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;

/**
 * 生产者
 */
@Slf4j
@RestController
public class ProducerController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/send/{msg}")
    public String msg(@PathVariable("msg") String msg) {
        log.info("当前时间：{},发送一条信息给两个 TTL 队列:{}", LocalDateTime.now(), msg);
        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_X, RabbitmqConfig.ROUTING_KEY_XA, "消息来自 ttl 为 10S 的队列: " + msg);
        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_X, RabbitmqConfig.ROUTING_KEY_XB, "消息来自 ttl 为 40s 的队列: " + msg);
        return "发送消息成功";
    }

}

消费者RabbitmqListener.java:

import com.github.config.RabbitmqConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;

/**
 * 消费者
 */
@Slf4j
@Component
public class RabbitmqListener {

    @RabbitListener(queues = RabbitmqConfig.DEAD_QUEUE_D)
    public void receive(Message message) {
        log.info("当前时间：{},收到死信队列信息：{}", LocalDateTime.now(), new String(message.getBody(), StandardCharsets.UTF_8));
    }

}

延迟队列优化

上面使用TTL实现了延迟队列，但是此时有些问题，如果现在我需要 5 min、10 min……，那么我岂不是每增加一个时间需求，就需要增加一个队列，如果是预定会议提前通知的场景，难道要增加无数个队列来满足要求？

解决：在消费者那边设置消息的 TTL 时间。

但是注意： RabbitMQ只会检查队列头部的消息是否过期，如果过期就放到死信队列，假如第一个过期时间很长，10s，第二个消息3s，则系统先看第一个消息，等到第一个消息过期，放到DLX。此时才会检查第二个消息，但实际上此时第二个消息早已经过期了，但是并没有先于第一个消息放到DLX。 使用插件不会出现这个问题，所以推荐使用插件实现延迟队列。

本图文内容来源于网友网络收集整理提供，作为学习参考使用，版权属于原作者。
THE END
分享
二维码
< <上一篇
下一篇>>