当前位置： 首页 > article >正文

【RabbitMQ】死信队列、延迟队列

article 2024/11/13 18:17:10

死信队列

死信，简单理解就是因为种种原因，无法被消费的消息。

有死信，自然就有死信队列。当一个消息在一个队列中变成死信消息之后，就会被重新发送到另一个交换器中，这个交换器就是DLX（Dead Letter Exchange），绑定该交换器的队列，就被称为死信队列DLQ（Dead Letter Queue）。

消息变成死信消息一般是由于以下几条：

  • 队列达到最大长度
  • 消息过期
  • 消息被拒绝，即消息确认机中手动确认的两种拒绝情况，并且不允许重新入队

队列达到最大长度

spring:
  rabbitmq:
    host: 43.138.108.125
    port: 5672
    username: admin
    password: admin
    virtual-host: mq-springboot-test
@Configuration
public class DeadConfig {

    // 正常队列，当正常队列中的消息出现一些不确定情况时，消息就会进入死信交换机中

    @Bean("normalQueue")
    public Queue normalQueue() {
        return QueueBuilder.durable(Constants.NORMAL_QUEUE)
                .deadLetterExchange(Constants.DEAD_EXCHANGE) // 设置死信交换机
                .deadLetterRoutingKey("dead") // 设置死信队列的路由键为dead
                .maxLength(3) // 设置队列的最大长度为3
                .build();
    }

    @Bean("normalExchange")
    public Exchange normalExchange() {
        return ExchangeBuilder.directExchange(Constants.NORMAL_EXCHANGE).durable(true).build();
    }

    @Bean("normalQueueBind")
    public Binding normalQueueBind(@Qualifier("normalQueue") Queue queue,
                                   @Qualifier("normalExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("normal").noargs();
    }

    // 死信队列

    @Bean("deadQueue")
    public Queue deadQueue() {
        return QueueBuilder.durable(Constants.DEAD_QUEUE).build();
    }

    @Bean("deadExchange")
    public Exchange deadExchange() {
        return ExchangeBuilder.directExchange(Constants.DEAD_EXCHANGE).durable(true).build();
    }

    @Bean("deadQueueBind")
    public Binding deadQueueBind(@Qualifier("deadQueue") Queue queue,
                                 @Qualifier("deadExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("dead").noargs();
    }

}
@RestController
@RequestMapping("/dead")
public class DeadController {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @RequestMapping
    public void deadQueue() {
        this.rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE, "normal", "hello 死信");
        System.out.println("正常队列发送消息成功");
    }

}
@Configuration
public class DeadListener {

    @RabbitListener(queues = Constants.DEAD_QUEUE)
    public void deadListener(String msg) {
        System.out.println("死信队列接收到消息：" + msg);
    }

}

在上述代码中，主要内容是声明了正常队列、交换机和绑定关系以及声明死信队列、死信交换机以及其绑定关系、正常队列的生产者代码、死亡队列的消费者代码。

队列达到最大长度和死信消息要转发到的DLX和路由键都是由正常队列在声明时进行绑定的。

启动上述程序之后，当正常队列存在三条消息之时，假设再来消息，那么消息就要进入死信交换机，从而路由到死信队列了。如下图可以看出，当发送第四条消息之后，死信队列的消费者就消费了一条消息：

在上述图片中，D表示队列是持久化的，Lim表示队列有最大长度，DLX表示队列存在死信交换机、DLK表示队列存在路由键。把鼠标放在这些字母上方，详细的消息都会表示。

在下述代码中，主要是对上述代码改进之后地方的指出，并没有把所有的代码全部给出。

消息过期

消息过期分为两种，一种是设置队列过期时间让消息过期，另一种是设置消息过期时间让消息过期，都可以进行测试。

设置队列过期时间

    @Bean("normalQueue")
    public Queue normalQueue() {
        return QueueBuilder.durable(Constants.NORMAL_QUEUE)
                .deadLetterExchange(Constants.DEAD_EXCHANGE) // 设置死信交换机
                .deadLetterRoutingKey("dead") // 设置死信队列的路由键为dead
//                .maxLength(3) // 设置队列的最大长度为3
                .ttl(5 * 1000) // 设置队列的过期时间为5秒
                .build();
    }

 由上图以及结合代码可以看出，将消息由正常生产者发送给Broker之后，大概5秒钟之后，消息过期。此时消息就会发送给死信交换机，从而交给其对应的消费者消费。

设置消息的过期时间

@Slf4j
@RestController
@RequestMapping("/dead")
public class DeadController {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @RequestMapping
    public void deadQueue() {
        // 设置消息的过期时间
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("5000");
                return message;
            }
        };
        this.rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE, "normal", "hello 死信", messagePostProcessor);
        log.info("死信队列发送成功");
    }

}

同样，结合上图和代码来说，19秒的时候消息发送功，24秒的时候死信消费者消费消息成功。

消息被拒绝

spring:
  rabbitmq:
    host: 43.138.108.125
    port: 5672
    username: admin
    password: admin
    virtual-host: mq-springboot-test
    listener:
      simple:
        acknowledge-mode: manual # 消息确认机制，手动确认
@Slf4j
@Configuration
public class DeadListener {

    // 正常队列接收消息
    @RabbitListener(queues = Constants.NORMAL_QUEUE)
    public void normalListener(Channel channel, String msg, Message message) throws IOException {
        try {
            log.info("正常队列监听器接收消息：{}", msg);
            int num = 3 / 0;
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        } catch (Exception e) {
            log.error("正常队列监听器接收消息异常：{}", e.getMessage());
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    // 死信队列接收消息
    @RabbitListener(queues = Constants.DEAD_QUEUE)
    public void deadListener(String msg, Channel channel, Message message) throws IOException {
        try {
            log.info("死信队列监听器接收消息：{}", msg);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        } catch (Exception e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
        }
    }

}

由上图以及代码可以看到，当消息的确认机制是手动确认时，当出现异常并且拒绝消息重新入队以后，消息就会来到死信队列中。

使用场景

用户支付订单之后，支付系统会给订单系统返回当前订单的支付状态。为了保证支付信息不丢失，需要使用到死信队列机制。当消息消费异常时，将消息投入到死信队列，由订单系统的其他消费者来监听这个队列，并对数据进行处理（比如发送工单等，进行人工确认）。

消息重试：将死信消息发送到原队列或另一个队列进行重试处理。

消息丢弃：直接丢弃这些无法处理的消息，避免占用系统资源。

日志收集：将死信消息做为日志收集起来，用户后续分析和问题定位。

延迟队列

概念

延迟队列就是消息发送之后，并不想让消费者立即拿到消息，而是在等待特定时间之后，消费者才能拿到消息进行消费

应用场景

  1. 用户发起退款后，24小时内商家未处理，默认退款
  2. 用户注册成功后，三天后发送短信，提高用户活跃度
  3. 预定会议后，在会议开始前15分钟提醒众人参加会议
  4. 用户通过手机远程遥控家里的智能设备在指定时间进行工作，这就可以使用延迟队列。用户发送消息到延迟队列，当指定时间到了再将指令推送到智能设备。

实现方法

  1. RabbitMQ本身并没有实现延迟队列，因此可以使用TTL + 死信队列的方式来实现延迟队列。
  2. 安装延迟队列插件来实现延迟队列。

TTL + 死信队列

@Configuration
public class MockDelayConfig {

    @Bean("mockDelayNormalQueue")
    public Queue mockDelayNormalQueue() {
        return QueueBuilder.durable(Constants.MOCk_DELAY_NORMAL_QUEUE)
                .ttl(5000 * 10) // 设置消息过期时间为50秒
                .deadLetterExchange(Constants.MOCK_DELAY_DEAD_EXCHANGE) // 设置死信交换机
                .deadLetterRoutingKey("mock.delay.dead") // 设置死信路由键
                .build();
    }

    @Bean("mockDelayNormalExchange")
    public Exchange mockDelayNormalExchange() {
        return ExchangeBuilder.directExchange(Constants.MOCk_DELAY_NORMAL_EXCHANGE).durable(true).build();
    }

    @Bean("mockDelayNormalQueueBind")
    public Binding mockDelayNormalQueueBind(@Qualifier("mockDelayNormalQueue") Queue queue,
                                           @Qualifier("mockDelayNormalExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("mock.delay.normal").noargs();
    }

    @Bean("mockDelayDeadQueue")
    public Queue mockDelayDeadQueue() {
        return QueueBuilder.durable(Constants.MOCK_DELAY_DEAD_QUEUE).build();
    }

    @Bean("mockDelayDeadExchange")
    public Exchange mockDelayDeadExchange() {
        return ExchangeBuilder.directExchange(Constants.MOCK_DELAY_DEAD_EXCHANGE).durable(true).build();
    }

    @Bean("mockDelayDeadQueueBind")
    public Binding mockDelayDeadQueueBind(@Qualifier("mockDelayDeadQueue") Queue queue,
                                         @Qualifier("mockDelayDeadExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("mock.delay.dead").noargs();
    }

}
@Slf4j
@RestController
@RequestMapping("/mockDelay")
public class MockDelayController {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @RequestMapping
    public void mockDelayQueue() {
        this.rabbitTemplate.convertAndSend(Constants.MOCk_DELAY_NORMAL_EXCHANGE,
                "mock.delay.normal", "hello 延迟队列");
        log.info("延迟队列生产者发送成功");
    }

}
@Slf4j
@Configuration
public class MockDelayListener {

    @RabbitListener(queues = Constants.MOCK_DELAY_DEAD_QUEUE)
    public void mockDelayListener(String msg) {
        log.info("模拟延迟队列消费者接收到消息：" + msg);
    }

}

在上述代码中，实现的功能是生产者发送消息后，消费者在50秒之后获得消息，对消息进行消费：

在TTL一文中，已经说明了RabbitMQ只会检查队首消息是否过期，不会扫描整个队列。因此如果想要放在模拟延迟队列中的消息过期时间不一致，那就会出现死信消息无法被及时处理的情况。因此，我们想要模拟实现延迟队列，就要确保队列中所有消息的过期时间是一致的。如果存在时间不一致的情况，我们就可以使用不同的模拟延迟队列来实现。

延迟队列插件

下载插件：官方网站进行下载（注意版本对应关系）

启动插件

rabbitma-plusins list // 查看插件列表

rabbitmq-plugins enable rabbitmq_delayed_message_exchange // 启动插件

service rabbitmq-server restart # 重启服务

如下图，当交换机中有了x-delayed-message就表示延迟插件安装成功 

代码测试

@Configuration
public class DelayConfig {

    @Bean("delayQueue")
    public Queue delayQueue() {
        return QueueBuilder.durable(Constants.DELAY_QUEUE).build();
    }

    @Bean("delayExchange")
    public Exchange delayExchange() {
        return ExchangeBuilder.directExchange(Constants.DELAY_EXCHANGE)
                .delayed() // 延迟交换机
                .durable(true) // 持久化
                .build();
    }

    @Bean("delayQueueBind")
    public Binding delayQueueBind(@Qualifier("delayQueue") Queue delayQueue,
                                   @Qualifier("delayExchange") Exchange delayExchange) {
        return BindingBuilder.bind(delayQueue).to(delayExchange).with("delay").noargs();
    }

}
@Slf4j
@RestController
@RequestMapping("/delay")
public class DelayController {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @RequestMapping
    public void delayQueue() {
        for(int i = 0; i < 5; i++) {
            // 随机生成延迟时间
            Random random = new Random();
            int time = random.nextInt(20);
            // 消息处理器，设置延迟时间
            MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setDelayLong((long) (time * 1000)); // 设置延迟时间
                    return message;
                }
            };
            // 发送消息到延迟队列
            this.rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, "delay", "hello 延迟队列 " + i, messagePostProcessor);
            log.info("发送延迟队列第" + i + "消息成功，延迟时间为：" + time);
        }
    }

}
@Slf4j
@Configuration
public class DelayListener {

    @RabbitListener(queues = Constants.DELAY_QUEUE)
    public void delayListener(String msg) {
        log.info("延迟队列监听器，接收到的消息：{}", msg);
    }

}

本质上，延迟插件就是让消息停留在交换机中，等到延迟时间结束之后，再发送到对应的队列中去。 

两者对比

使用TTL + 死信队列的好处是不需要额外安装插件。缺点是受消息的延迟时间影响，同一个队列中的消息必须延迟时间相同。

使用延迟队列插件的好处是不受延迟时间影响，同一队列中的所有消息延迟时间可以不同，额外的插件使得延迟队列的实现比较容易。缺点是需要依赖特定的插件，并且插件的版本必须和对应的RabbitMQ相对应。


http://www.kler.cn/a/313460.html

相关文章：

  • 程序员年薪百万秘籍（一）
  • LLM时代下Embedding模型如何重塑检索、增强生成
  • btstack协议栈实战篇--SDP Client - Query Remote SDP Records
  • 速盾：游戏盾的功能和原理详解
  • 一篇Spring Boot 笔记
  • C# 软件测试
  • windows下用cmake编译腾讯云的对象存储COS的XML C++SDK
  • java通过org.eclipse.milo实现OPCUA客户端进行连接和订阅
  • 2-93 基于matlab的无人机FMCW（频率调制连续波）毫米波高度计雷达仿真
  • axios(基于Promise的HTTP客户端) 与 `async` 和 `await` 结合使用
  • 中级练习[5]：Hive SQL用户行为与商品价格综合分析
  • Docker笔记-容器数据卷
  • 做到三点从“穷人思维”转变为“富人思维”
  • Node-red 某一时间范围内满足条件的数据只返回一次
  • 前端univer创建、编辑excel
  • 大模型爬虫—ScrapeGraphAI
  • AutoDL云计算GPT-SoVITS-TTS语音声色克隆语音合成
  • 卡牌抽卡机小程序：市场发展下的创新
  • 【Webpack--006】处理字体图标资源
  • 【2024华为杯数学建模竞赛】E题 解题思路 | 视频特征提取
  • 管理和合并多个开发者的 Git 提交：团队协作的艺术
  • 2023-基于深度学习的射频指纹识别与信道效应缓解
  • 使用Django 搭建自动化平台
  • 基于JavaWeb开发的java springboot+mybatis电影售票网站管理系统前台+后台设计和实现
  • Angular 2 TypeScript：现代Web开发的利器
  • Java免税购物商城：Spring Boot技术实现