当前位置:   article > 正文

RabbitMQ学习(八):死信队列_rabbitmq死信队列实战

rabbitmq死信队列实战

一、死信队列概述

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

有死信自然也有死信队列。

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

二、死信的原因

  • 消息 TTL 过期

  • 队列达到最大长度(队列满了,无法再添加数据)

  • 消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false

三、死信实战(TTL过期)

代码架构图

正常情况下,消息会进入普通队列,由C1消费;而当成为死信之后,会进入死信队列,由C2消费。

生产者核心代码:

  1. //声明普通交换机
  2. channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
  3. //设置消息的 TTL 时间
  4. AMQP.BasicProperties properties =
  5.     new AMQP.BasicProperties().builder().expiration("10000").build();
  6. for (int i = 1; i <11 ; i++) {
  7.     String message="info"+i;
  8.     channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
  9. }

消费者C1核心代码:

  1. //声明死信和普通交换机 类型为 direct
  2. channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
  3. channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
  4. //声明死信队列
  5. channel.queueDeclare("dead-queue", false, false, false, null);
  6. //死信队列绑定死信交换机与 routingkey
  7. channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
  8. //正常队列绑定死信队列信息
  9. Map<String, Object> params = new HashMap<>();
  10. //正常队列设置死信交换机 参数 key 是固定值
  11. params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
  12. //正常队列设置死信 routing-key 参数 key 是固定值
  13. params.put("x-dead-letter-routing-key", "lisi");
  14. channel.queueDeclare("normal-queue", false, false, false, params);
  15. channel.queueBind("normal-queue", NORMAL_EXCHANGE, "zhangsan");
  16. DeliverCallback deliverCallback = (consumerTag, delivery) -> {
  17.     String message = new String(delivery.getBody(), "UTF-8");
  18. };
  19. channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {});

消费者C2核心代码:

  1. //声明死信队列
  2. channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
  3. channel.queueDeclare("dead-queue", false, false, false, null);
  4. channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
  5. DeliverCallback deliverCallback = (consumerTag, delivery) -> {
  6.     String message = new String(delivery.getBody(), "UTF-8");
  7. };
  8. channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {});

代码略微改动即可模拟其他两个原因下的死信。

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

闽ICP备14008679号