六、死信队列

简介: 六、死信队列

📄 死信队列的概念:

死信就是无法被消费的消息。一般来说,producer 将消息投递到 broker 或者直接到 queue 中,consumer 从 queue 取出消息进行消费,但某些时候由于特定的原因导致 queue 中的某些消息无法被消费,这样的消息如果没有后续的处理,就变成了死信,有死信自然就有了死信队列。

应用场景:

为了保证订单业务的消息数据不丢失,需要使用到 RabbitMQ 的死信队列机制,当消息消费发生异常时,将消息投入死信队列中。
还有比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

死信的原因:

  • 消息 TTL 过期
  • 队列达到最大长度(队列满了无法再添加数据到 mq 中)
  • 消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false
三种导致死信的效果:消息被拒绝、消息TTL过期、队列达到最大长度。架构图如下所示:

1. 消息TTL过期 , 代码示例

(1) application.yaml

# 项目名
server:
  servlet:
    context-path: /demo
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    publisher-confirm-type: correlated # 开启消息确认
    publisher-returns: true # 开启发送失败回退
    template:
      mandatory: true
    listener:
      simple:
        acknowledge-mode: manual  #消息手动确认
        concurrency: 1 #消费者数量
        max-concurrency: 1  #消费者最大数量
        prefetch: 1  #消费者每次从队列中取几个消息
        default-requeue-rejected: true #消息消费失败后,重新进入消费队列中
        retry:
          initial-interval: 1000 #1秒后重试
          enabled: true #启用发布重试
          max-attempts: 3 #传递消息的最大尝试次数
          max-interval: 10000 #尝试的最大时间间隔
          multiplier: 1.0 #应用于先前传递重试时间间隔的乘数

(2) RabbitConfig

@Configuration
public class RabbitConfig implements BeanPostProcessor {
    //1. 创建交换机
    @Bean
    public DirectExchange newExchange(){
        return new DirectExchange("normalExchange",true,false);
    }
    //2. 创建队列
    @Bean
    public Queue newQueue(){
        Map<String ,Object> map = new HashMap<>();
        map.put("x-message-ttl",2000); // 消息存活时间1s
        map.put("x-dead-letter-exchange","deadExchange");// 设置死信交换机 的名称
        map.put("x-dead-letter-routing-key","key2") ;//设置死信路由键名字
        return new Queue("normalQueueA",true,false,false,map);
    }
    //3. 绑定
    @Bean
    public Binding binding(){
        return BindingBuilder.bind(newQueue()).to(newExchange()).with("key1");
    }
    //4. 创建死信交换机
    @Bean
    public DirectExchange newDeadExchange(){
        return new DirectExchange("deadExchange",true,false);
    }
    //5. 创建死信队列
    @Bean
    public Queue newDeadQueue(){
        return new Queue("deadQueueA",true,false,false);
    }
    //6. 绑定
    @Bean
    public Binding bindingDead(){
        return BindingBuilder.bind(newDeadQueue()).to(newDeadExchange()).with("key2");
    }
}

(3) 生产者:订单发送消息给下游服务 支付服务

@RestController
public class OrderProducer {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    @PostMapping("/submitOrder")
    //模拟下订单
    public String submitOrder(){
        String orderId = UUID.randomUUID().toString().replace("-","");
        Map<String ,Object> map = new HashMap<>();
        map.put("orderId",orderId);
        map.put("orderNum",4959044);
        map.put("custId",101);
        map.put("orderPrice",3500f);
        map.put("orderDate",new Date());
        rabbitTemplate.convertAndSend("normalExchange","key1",map);
        return "生产者下单成功";
    }
}

(4)下游服务消费者:支付服务

@Component
public class PayConsumer {
        @RabbitHandler
       @RabbitListener(queues = "normalQueueA")
    public void process(Map map,Channel channel,Message message) throws IOException {
        System.out.println("支付服务接收到的消息:" + map);
        String orderId = (String)map.get("orderId");
        Integer custId = (Integer)map.get("custId");
        Integer orderNum = (Integer)map.get("orderNum");
        Float orderPrice = (Float)map.get("orderPrice");
        Date orderDate = (Date)map.get("orderDate");
        System.out.println("支付服务接收到的orderId:" + orderId);
        System.out.println("支付服务接收到的custId:" + custId);
        System.out.println("支付服务接收到的orderNum:" + orderNum);
        System.out.println("支付服务接收到的orderPrice:" + orderPrice);
        System.out.println("支付服务接收到的orderDate:" + orderDate);
        //告诉broker,消息已经被确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}

(5)消息未处理后,消费死信消息的消费者

@Component
public class DeadOrderConsumer {
    // 获得死信队列中的消息
     @RabbitHandler
   @RabbitListener(queues = "deadQueueA")
    public void process(Map map){
        System.out.println("订单取消支付后,从死信队列中接收到的消息:" + map);
        String orderId = (String)map.get("orderId");
        Integer custId = (Integer)map.get("custId");
        Integer orderNum = (Integer)map.get("orderNum");
        Float orderPrice = (Float)map.get("orderPrice");
        Date orderDate = (Date)map.get("orderDate");
        System.out.println("取消支付后,从死信队列中接收到的orderId:" + orderId);
        System.out.println("取消支付后,从死信队列中接收到的custId:" + custId);
        System.out.println("取消支付后,从死信队列中接收到的orderNum:" + orderNum);
        System.out.println("取消支付后,从死信队列中接收到的orderPrice:" + orderPrice);
        System.out.println("取消支付后,从死信队列中接收到的orderDate:" + orderDate);
    }
}

若将PayConsumer中的注解注释了,或监听其他不存在的queue, 即 支付服务 没有消费 订单消息,则订单发送发送过来的消息 过了ttl 时间,则进入死信队列。运行效果如下:

2. 队列达到最大长度

application.yaml 不变

RabbitConfig 去掉ttl 属性, 加上 队列达到最大长度 为6

//2. 创建队列
    @Bean
    public Queue newQueue(){
        Map<String ,Object> map = new HashMap<>();
        map.put("x-max-length",6); // 队列达到最大长度 为6
        map.put("x-dead-letter-exchange","deadExchange");// 设置死信交换机 的名称
        map.put("x-dead-letter-routing-key","key2") ;//设置死信路由键名字
        return new Queue("normalQueueA",true,false,false,map);
    }

OrderProducer 循环发送10个消息

@RestController
public class OrderProducer {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    @PostMapping("/submitOrder")
    //模拟下订单
    public String submitOrder(){
        Map<String ,Object> map = new HashMap<>();      
        map.put("orderNum",4959044);
        map.put("custId",101);
        map.put("orderPrice",3500f);
        map.put("orderDate",new Date());
        for(int i =0;i<=9;i++) {
            String orderId = UUID.randomUUID().toString().replace("-", "");
            map.put("orderId",orderId);
            rabbitTemplate.convertAndSend("normalExchange", "key1", map);
        }
        return "生产者下单成功";
    }
}

下游消费者 支付服务PayConsumer, DeadOrderConsumer 不变

进行测试,首先启动 Consumer01 创建更改后的队列,然后关闭它,模拟其接受不到消息。然后启动消费者发送10条消息,可以看到发送到6条消息首先在 normal-queue 队列中,4条消息在 dead-queue 死信队列中。

目录
相关文章
|
消息中间件 存储 Java
RabbitMQ之延迟队列解读
RabbitMQ之延迟队列解读
|
8月前
|
Shell API
最新,通过GACCode配置Codex桌面客户端使用方案!
本指南详解Codex桌面端配置流程:①安装客户端;②获取gaccode.com的API Key(无订阅可邮件申请);③创建并编辑~/.codex/config.toml文件,配置gac模型提供方与gpt-5.4模型;④导出CODEX_API_KEY环境变量;⑤重启应用即可开始智能对话。
最新,通过GACCode配置Codex桌面客户端使用方案!
|
11月前
|
JSON Java 数据格式
Spring Boot中的全局异常处理
本文介绍了Spring Boot项目中如何实现全局异常处理。通过@ControllerAdvice和@ExceptionHandler统一捕获系统异常与自定义业务异常,结合统一JSON返回结构,避免代码耦合,提升可维护性。同时利用枚举管理异常码,实现异常信息的集中管理和友好提示,适用于前后端分离及微服务架构,保障接口返回一致性,降低维护成本。
|
11月前
|
监控 前端开发 Java
Spring Boot 统一异常处理与全局响应增强
本文介绍如何在Spring Boot中实现全局异常处理,通过`@RestControllerAdvice`和自定义`BusinessException`统一拦截异常,结合`JsonResult`标准化响应格式,避免信息泄露,提升前后端协作效率与系统健壮性。
|
12月前
|
SQL 缓存 Java
【Java架构必看】Mybatis的工作原理
MyBatis执行分启动与运行两阶段:启动时加载配置,运行时代理执行SQL。通过JDK动态代理生成Mapper接口,结合缓存机制与Executor执行SQL,最终由TypeHandler完成结果映射。
638 4
|
11月前
|
XML Java 数据库连接
Spring Boot集成MyBatis
MyBatis 是一款优秀的持久层框架,支持SQL映射与注解两种方式,简化数据库操作。本文详解Spring Boot集成MyBatis的配置方法,涵盖依赖引入、YAML配置、XML与注解模式整合,并重点讲解@MapperScan、@Param、@Results等注解使用技巧,助力开发者高效实现数据访问。
2025版基于springboot的校园打印社管理系统
本系统旨在解决传统校园打印社管理效率低、排队时间长、耗材管理难等问题,集成订单管理、设备监控、耗材统计、线上预约、自助打印与在线支付等功能,提升运营效率与用户体验,助力校园信息化建设。
|
存储 人工智能 缓存
AI助理直击要害,从繁复中提炼精华——使用CDN加速访问OSS存储的图片
本案例介绍如何利用AI助理快速实现OSS存储的图片接入CDN,以加速图片访问。通过AI助理提炼关键操作步骤,避免在复杂文档中寻找解决方案。主要步骤包括开通CDN、添加加速域名、配置CNAME等。实测显示,接入CDN后图片加载时间显著缩短,验证了加速效果。此方法大幅提高了操作效率,降低了学习成本。
6410 16
|
机器学习/深度学习 存储 安全
HashMap源码详解,太顶了这次
HashMap源码详解,太顶了这次
HashMap源码详解,太顶了这次
|
Java 数据库连接 数据库
【Spring技术专题】「实战开发系列」保姆级教你SpringBoot整合Mybatis框架实现多数据源的静态数据源和动态数据源配置落地
Mybatis是一个基于JDBC实现的,支持普通 SQL 查询、存储过程和高级映射的优秀持久层框架,去掉了几乎所有的 JDBC 代码和参数的手工设置以及对结果集的检索封装。 Mybatis主要思想是将程序中大量的 SQL 语句剥离出来,配置在配置文件中,以实现 SQL 的灵活配置。在所有 ORM 框架中都有一个非常重要的媒介——PO(持久化对象),PO 的作用就是完成持久化操作,通过该对象对数据库执行增删改的操作,以面向对象的方式操作数据库。
677 1
【Spring技术专题】「实战开发系列」保姆级教你SpringBoot整合Mybatis框架实现多数据源的静态数据源和动态数据源配置落地

热门文章

最新文章