devbox
花生_TL007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ——死信队列介绍和项目应用_rabbitmq 订单系统死信队列的应用

作者：花生_TL007 | 2024-05-14 19:07:59

赞

踩

rabbitmq 订单系统死信队列的应用

前言

最近在做一个BI项目，用到了RabbitMQ异步化生成图表，同时还添加了死信队列处理无法被消费者正常消费的消息。于是便有了这篇文章，下面由我带大家介绍RabbitMQ的死信队列和其在项目中的应用吧。

死信和死信队列的概念

什么是死信？简单来说就是无法被消费和处理的消息。一般生产者将消息投递到broker或者queue，消费者直接从中取出消息进行消费。但有时因为某些原因导致消息不能被消费，导致消息积压在队列中，这样的消息如果没有后续的处理就会变成死信，那么专门存放死信的队列就是死信队列。

注意：如果一个消息队列设置了过期时间，在队列过期后其中的消息并不会自动转发到死信队列中，而是会被系统丢弃或执行其他的操作。

什么是死信交换机？

那么什么是死信交换机呢？死信交换机是指专门将死信路由到死信队列的交换机。

注意：死信交换机和普通交换机无任何区别，只是路由的消息类别不同。同理，死信队列和普通队列也无任何区别，仅仅是一个专门存放死信的队列而已。无论是死信交换机还是队列，它们的创建方法都和普通的交换机和队列一致，无非是名称和routingKey不同。

产生死信的原因

根据官方文档，我们发现一般有三种场景会产生死信。

  1. 消息超过TTL，即消息过期
  2. 消息被nack或reject，且不予重新入队
  3. 队列达到最大长度

死信队列实战和应用

死信队列的应用并不难，无非就是多定义了一个交换机、routingKey和队列罢了。在声明普通队列时传入Map参数，往Map中put死信队列名称、死信routingKey、消息TTL等参数即可完成死信自动投递到死信队列的流程。通过如下代码即可绑定普通队列和死信交换机了，而且还能设置routingKey和队列长度等参数，无需像传统的那样通过channel绑定。

  1. ​
  2.         Map<String, Object> arguments = new HashMap<>();
  3.         // 过期时间
  4.         arguments.put("x-message-ttl", 10000);
  5.         // 正常队列设置死信交换机
  6.         arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
  7.         // 设置死信routingKey
  8.         arguments.put("x-dead-letter-routing-key", "lisi");
  9.         // 设置正常队列的长度限制
  10.         arguments.put("x-max-length", 10);
  11.  
  12. ​

流程图：

 

生产者Producer：

  1. public class Producer {
  2.     // 普通交换机名称
  3.     public static final String NORMAL_EXCHANGE = "normal_exchange";
  4.  
  5.     public static void main(String[] args) throws IOException {
  6.         Channel channel = RabbitMQUtils.getChannel();
  7.         //死信消息 设置TTL时间
  8.         AMQP.BasicProperties properties = new AMQP.BasicProperties()
  9.                 .builder().expiration("10000").build();
  10.  
  11.         // 延迟消息
  12.         for (int i = 0;i < 10;i++) {
  13.             String message = i + "info";
  14.             channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
  15.         }
  16.     }
  17. }
  18.  
  19. ​

普通队列消费者C1：

  1. public class Consumer01 {
  2.     // 普通交换机名称
  3.     public static final String NORMAL_EXCHANGE = "normal_exchange";
  4.     // 死信交换机名称
  5.     public static final String DEAD_EXCHANGE = "dead_exchange";
  6.     // 普通队列名称
  7.     public static final String NORMAL_QUEUE = "normal_queue";
  8.     // 死信队列名称
  9.     public static final String DEAD_QUEUE = "dead_queue";
  10.  
  11.     public static void main(String[] args) throws IOException {
  12.         Channel channel = RabbitMQUtils.getChannel();
  13.         // 声明死信和普通交换机，类型为direct
  14.         channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
  15.         channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
  16.         // 声明普通队列
  17.         Map<String, Object> arguments = new HashMap<>();
  18.         // 过期时间
  19.         arguments.put("x-message-ttl", 10000);
  20.         // 正常队列设置死信交换机
  21.         arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
  22.         // 设置死信routingKey
  23.         arguments.put("x-dead-letter-routing-key", "lisi");
  24.         // 设置正常队列的长度限制
  25.         arguments.put("x-max-length", 10);
  26.  
  27.         // 声明普通队列
  28.         channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
  29.         // 声明死信队列
  30.         channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
  31.  
  32.         channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
  33.         channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");
  34.         System.out.println("consumer01等待接收消息");
  35.  
  36.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  37.             String msg = new String(message.getBody(), "UTF-8");
  38.             if (msg.equals("info5")) {
  39.                 System.out.println("consumer01接收的消息：" + new String(message.getBody()));
  40.                 System.out.println(msg + "：此消息是被拒绝的");
  41.                 channel.basicReject(message.getEnvelope().getDeliveryTag(), false); //拒绝此消息并不放回普通队列
  42.             } else {
  43.                 System.out.println("consumer01接收的消息：" + new String(message.getBody()));
  44.                 channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
  45.             }
  46.         };
  47.  
  48.         CancelCallback cancelCallback = consumerTag -> {
  49.             System.out.println("C1取消消息");
  50.         };
  51.         channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, cancelCallback);
  52.     }
  53. }

死信队列消费者C2：

  1. public class Consumer02 {
  2.     // 死信队列名称
  3.     public static final String DEAD_QUEUE = "dead_queue";
  4.  
  5.     public static void main(String[] args) throws IOException {
  6.         Channel channel = RabbitMQUtils.getChannel();
  7.         System.out.println("consumer02等待接收消息");
  8.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  9.             System.out.println("consumer02接收的消息：" + new String(message.getBody()));
  10.         };
  11.  
  12.         CancelCallback cancelCallback = consumerTag -> {
  13.             System.out.println("C2取消消息");
  14.         };
  15.         channel.basicConsume(DEAD_QUEUE, true, deliverCallback, cancelCallback);
  16.     }
  17. }

依次启动生产者，和两个消费者，并停掉普通队列的消费者，我们发现生产者发送的消息被死信队列消费者C2给接收了。

在上面的代码中，我在普通队列中设置了消息的TTL为5s，但是我又在生产者设置发送的消息TTL为10s，那么RabbitMQ会以哪个为准呢？其实RabbitMQ会以较短的TTL为准

BI项目添加死信队列

声明交换机、队列和routingKey的配置类

  1. @Configuration
  2. public class TtlQueueConfig {
  3.     private final String COMMON_EXCHANGE = "bi_common_exchange"; // 普通交换机名称
  4.     private final String COMMON_QUEUE = "bi_common_queue"; // 普通队列名称
  5.     private final String DEAD_LETTER_EXCHANGE = "bi_dead_letter_exchange"; // 死信交换机名称
  6.     private final String DEAD_LETTER_QUEUE = "bi_dead_letter_queue"; // 死信队列名称
  7.     private final String COMMON_ROUTINGKEY = "bi_common_routingKey"; // 普通routingKey
  8.     private final String DEAD_LETTER_ROUTINGKEY = "bi_dead_letter_routingKey"; // 死信routingKey
  9.  
  10.     // 普通交换机
  11.     @Bean("commonExchange")
  12.     public DirectExchange commonExchange() {
  13.         return new DirectExchange(COMMON_EXCHANGE);
  14.     }
  15.  
  16.     // 死信交换机
  17.     @Bean("deadLetterExchange")
  18.     public DirectExchange deadLetterExchange() {
  19.         return new DirectExchange(DEAD_LETTER_EXCHANGE);
  20.     }
  21.  
  22.     // 普通队列
  23.     @Bean("commonQueue")
  24.     public Queue commonQueue() {
  25.         Map<String, Object> map = new HashMap<>(3);
  26.         map.put("x-message-ttl", 20000);
  27.         map.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
  28.         map.put("x-dead-letter-routing-key", DEAD_LETTER_ROUTINGKEY);
  29.         return QueueBuilder.durable(COMMON_QUEUE).withArguments(map).build();
  30.     }
  31.  
  32.     // 死信队列
  33.     @Bean("deadLetterQueue")
  34.     public Queue deadLetterQueue() {
  35.         return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
  36.     }
  37.  
  38.     @Bean
  39.     public Binding commonQueueBindingCommonExchange(@Qualifier("commonQueue") Queue commonQueue,
  40.                                                     @Qualifier("commonExchange") DirectExchange commonExchange) {
  41.         return BindingBuilder.bind(commonQueue).to(commonExchange).with(COMMON_ROUTINGKEY);
  42.     }
  43.  
  44.     @Bean
  45.     public Binding deadQueueBindingDeadExchange(@Qualifier("deadLetterQueue") Queue deadLetterQueue,
  46.                                                 @Qualifier("deadLetterExchange") DirectExchange deadLetterExchange){
  47.         return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with(DEAD_LETTER_ROUTINGKEY);
  48.     }
  49. }
  50.  
  51.  
  52. 普通消费者（负责异步生成图表）
  53.  
  54.  
  55. @Component
  56. @Slf4j
  57. public class BIMessageConsumer {
  58.     @Resource
  59.     private ChartService chartService;
  60.  
  61.     @Resource
  62.     private RabbitTemplate rabbitTemplate;
  63.  
  64.     @Resource
  65.     private AIManager aiManager;
  66.  
  67.     @Resource
  68.     RedisTemplate<String, Object> redisTemplate;
  69.  
  70.     // 制定消费者监听哪个队列和消息确认机制
  71.     @SneakyThrows
  72.     @RabbitListener(queues = {"bi_common_queue"}, ackMode = "MANUAL")
  73.     public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
  74.         log.info("receiveMessage is {}", message);
  75.         if(StringUtils.isBlank(message)) {
  76.             // 如果失败，消息拒绝
  77.             channel.basicNack(deliveryTag, false, false);
  78.             log.info("消息为空拒绝接收");
  79.             log.info("此消息正在被转发到死信队列中");
  80.         }
  81.  
  82.         long chartId = Long.parseLong(message);
  83.         Chart chart = chartService.getById(chartId);
  84.         if (chart == null) {
  85.             channel.basicNack(deliveryTag, false, false);
  86.             log.info("图标为空拒绝接收");
  87.             throw new BusinessException(ErrorCode.NOT_FOUND_ERROR, "图表为空");
  88.         }
  89.  
  90.         // 先修改图表任务状态为“执行中”。等执行成功后，修改为“已完成”、保存执行结果；执行失败后，状态修改为“失败”，记录任务失败信息。
  91.         Chart updateChart = new Chart();
  92.         updateChart.setId(chart.getId());
  93.         updateChart.setStatus("running");
  94.         boolean b = chartService.updateById(updateChart);
  95.         if (!b) {
  96.             channel.basicNack(deliveryTag, false, false);
  97.             handlerChartUpdateError(chart.getId(), "更新图表执行状态失败");
  98.             return;
  99.         }
  100.         // 调用AI
  101.         String result = aiManager.doChat(CommonConstant.BI_MODEL_ID, buildUserInput(chart));
  102.         String[] splits = result.split("【【【【【");
  103.         if (splits.length < 3) {
  104.             channel.basicNack(deliveryTag, false, false);
  105.             handlerChartUpdateError(chart.getId(), "AI生成错误");
  106.             return;
  107.         }
  108.         String genChart = splits[1].trim();
  109.         String genResult = splits[2].trim();
  110.         Chart updateChartResult = new Chart();
  111.         updateChartResult.setId(chart.getId());
  112.         updateChartResult.setGenChart(genChart);
  113.         updateChartResult.setGenResult(genResult);
  114.         updateChartResult.setStatus("succeed");
  115.         boolean updateResult = chartService.updateById(updateChartResult);
  116.         if (!updateResult) {
  117.             channel.basicNack(deliveryTag, false, false);
  118.             handlerChartUpdateError(chart.getId(), "更新图表成功状态失败");
  119.         }
  120.         Long userId = chartService.queryUserIdByChartId(chartId);
  121.         String myChartId = String.format("lingxibi:chart:list:%s", userId);
  122.         redisTemplate.delete(myChartId);
  123.  
  124.         // 如果任务执行成功，手动执行ack
  125.         channel.basicAck(deliveryTag, false);
  126.     }
  127.  
  128.  
  129.     private void handlerChartUpdateError(long chartId, String execMessage) {
  130.         Chart updateChartResult = new Chart();
  131.         updateChartResult.setId(chartId);
  132.         updateChartResult.setStatus("failed");
  133.         updateChartResult.setExecMessage(execMessage);
  134.         boolean updateResult = chartService.updateById(updateChartResult);
  135.         if (!updateResult) {
  136.             log.error("更新图表失败状态失败" + chartId + "," + execMessage);
  137.         }
  138.     }
  139.  
  140.     /**
  141.      * 构建用户输入
  142.      * @param chart
  143.      * @return
  144.      */
  145.     private String buildUserInput(Chart chart) {
  146.         String goal = chart.getGoal();
  147.         String chartType = chart.getChartType();
  148.         String csvData = chart.getChartData();
  149.  
  150.         // 构造用户输入
  151.         StringBuilder userInput = new StringBuilder();
  152.         userInput.append("分析需求：").append("\n");
  153.         // 拼接分析目标
  154.         String userGoal = goal;
  155.         if (StringUtils.isNotBlank(chartType)) {
  156.             userGoal += ",请使用" + chartType;
  157.         }
  158.         userInput.append(userGoal).append("\n");
  159.         userInput.append("原始数据：").append("\n");
  160.         // 压缩后的数据
  161.  
  162.         userInput.append(csvData).append("\n");
  163.         return userInput.toString();
  164.     }
  165. }

普通消费者（负责异步生成图表信息）

  1. @Component
  2. @Slf4j
  3. public class BIMessageConsumer {
  4.     @Resource
  5.     private ChartService chartService;
  6.  
  7.     @Resource
  8.     private RabbitTemplate rabbitTemplate;
  9.  
  10.     @Resource
  11.     private AIManager aiManager;
  12.  
  13.     @Resource
  14.     RedisTemplate<String, Object> redisTemplate;
  15.  
  16.     // 制定消费者监听哪个队列和消息确认机制
  17.     @SneakyThrows
  18.     @RabbitListener(queues = {"bi_common_queue"}, ackMode = "MANUAL")
  19.     public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
  20.         log.info("receiveMessage is {}", message);
  21.         if(StringUtils.isBlank(message)) {
  22.             // 如果失败，消息拒绝
  23.             channel.basicNack(deliveryTag, false, false);
  24.             log.info("消息为空拒绝接收");
  25.             log.info("此消息正在被转发到死信队列中");
  26.         }
  27.  
  28.         long chartId = Long.parseLong(message);
  29.         Chart chart = chartService.getById(chartId);
  30.         if (chart == null) {
  31.             channel.basicNack(deliveryTag, false, false);
  32.             log.info("图标为空拒绝接收");
  33.             throw new BusinessException(ErrorCode.NOT_FOUND_ERROR, "图表为空");
  34.         }
  35.  
  36.         // 先修改图表任务状态为“执行中”。等执行成功后，修改为“已完成”、保存执行结果；执行失败后，状态修改为“失败”，记录任务失败信息。
  37.         Chart updateChart = new Chart();
  38.         updateChart.setId(chart.getId());
  39.         updateChart.setStatus("running");
  40.         boolean b = chartService.updateById(updateChart);
  41.         if (!b) {
  42.             channel.basicNack(deliveryTag, false, false);
  43.             handlerChartUpdateError(chart.getId(), "更新图表执行状态失败");
  44.             return;
  45.         }
  46.         // 调用AI
  47.         String result = aiManager.doChat(CommonConstant.BI_MODEL_ID, buildUserInput(chart));
  48.         String[] splits = result.split("【【【【【");
  49.         if (splits.length < 3) {
  50.             channel.basicNack(deliveryTag, false, false);
  51.             handlerChartUpdateError(chart.getId(), "AI生成错误");
  52.             return;
  53.         }
  54.         String genChart = splits[1].trim();
  55.         String genResult = splits[2].trim();
  56.         Chart updateChartResult = new Chart();
  57.         updateChartResult.setId(chart.getId());
  58.         updateChartResult.setGenChart(genChart);
  59.         updateChartResult.setGenResult(genResult);
  60.         updateChartResult.setStatus("succeed");
  61.         boolean updateResult = chartService.updateById(updateChartResult);
  62.         if (!updateResult) {
  63.             channel.basicNack(deliveryTag, false, false);
  64.             handlerChartUpdateError(chart.getId(), "更新图表成功状态失败");
  65.         }
  66.         Long userId = chartService.queryUserIdByChartId(chartId);
  67.         String myChartId = String.format("lingxibi:chart:list:%s", userId);
  68.         redisTemplate.delete(myChartId);
  69.  
  70.         // 如果任务执行成功，手动执行ack
  71.         channel.basicAck(deliveryTag, false);
  72.     }
  73.  
  74.  
  75.     private void handlerChartUpdateError(long chartId, String execMessage) {
  76.         Chart updateChartResult = new Chart();
  77.         updateChartResult.setId(chartId);
  78.         updateChartResult.setStatus("failed");
  79.         updateChartResult.setExecMessage(execMessage);
  80.         boolean updateResult = chartService.updateById(updateChartResult);
  81.         if (!updateResult) {
  82.             log.error("更新图表失败状态失败" + chartId + "," + execMessage);
  83.         }
  84.     }
  85.  
  86.     /**
  87.      * 构建用户输入
  88.      * @param chart
  89.      * @return
  90.      */
  91.     private String buildUserInput(Chart chart) {
  92.         String goal = chart.getGoal();
  93.         String chartType = chart.getChartType();
  94.         String csvData = chart.getChartData();
  95.  
  96.         // 构造用户输入
  97.         StringBuilder userInput = new StringBuilder();
  98.         userInput.append("分析需求：").append("\n");
  99.         // 拼接分析目标
  100.         String userGoal = goal;
  101.         if (StringUtils.isNotBlank(chartType)) {
  102.             userGoal += ",请使用" + chartType;
  103.         }
  104.         userInput.append(userGoal).append("\n");
  105.         userInput.append("原始数据：").append("\n");
  106.         // 压缩后的数据
  107.  
  108.         userInput.append(csvData).append("\n");
  109.         return userInput.toString();
  110.     }
  111. }

死信队列消费者（负责处理死信）

收到死信后我是直接确认了，这种方式可能不好，你也可以换成其他方式比如重新入队，或者写入数据库并打上日志等等。

  1.  
  2. @Component
  3. @Slf4j
  4. public class TtlQueueConsumer {
  5.     @Resource
  6.     BIMessageProducer biMessageProducer;
  7.  
  8.     @SneakyThrows
  9.     @RabbitListener(queues = "bi_dead_letter_queue", ackMode = "MANUAL")
  10.     public void doTTLMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
  11.         log.info("已经接受到死信消息：{}", message);
  12.         biMessageProducer.sendMessage(message);
  13.         channel.basicAck(deliveryTag, false);
  14.     }
  15. }

如果我的文章对你有帮助的话，不妨给我点个赞呗，我会持续带来不一样的内容。如果对Java相关知识感兴趣的话，可以关注我，带你走进Java的世界。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/花生_TL007/article/detail/569684
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号