消息队列 — RabbitMQ06

6 死信队列 6.1 死信的概念

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

6.2. 死信的来源

消息TTL过期
队列达到最大长度(队列满了，无法再添加数据到mq中)
消息被拒绝(basic.reject或basic.nack)并且requeue=false

6.3. 死信实战 6.3.1 消息TTL过期

生产者代码

package com.zjw.rabbitmq.deadqueue.ttl;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.zjw.rabbitmq.util.RabbitMQUtil;


public class Producer {

  private static final String NORMAL_EXCHANGE = "normal_exchange";

  public static void main(String[] argv) throws Exception {
    try (Channel channel = RabbitMQUtil.getChannel()) {
      channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
      //设置消息的 TTL 时间 10s
      BasicProperties properties = new BasicProperties().builder().expiration("10000").build();

      for (int i = 0; i < 10; i++) {
        String message = "hello-" + i;
        channel.basicPublish(NORMAL_EXCHANGE, "InQueue", properties, message.getBytes());
        System.out.println("生产者发送消息:" + message);
      }
    }
  }
}

消费者 C1 代码(启动之后关闭该消费者模拟其接收不到消息)

package com.zjw.rabbitmq.deadqueue.ttl;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.zjw.rabbitmq.util.RabbitMQUtil;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;


public class Consumer01 {
  // 普通交换机名称
  private static final String NORMAL_EXCHANGE = "normal_exchange";
  // 死信交换机名称
  private static final String DEAD_EXCHANGE = "dead_exchange";
  // 普通队列
  private static final String NORMAL_QUEUE = "normal-queue";
  // 死信队列
  private static final String DEAD_QUEUE = "dead-queue";

  public static void main(String[] argv) throws Exception {
    Channel channel = RabbitMQUtil.getChannel();
    // 声明死信和普通交换机类型为direct
    channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
    channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
    // 声明死信队列
    channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
    // 死信队列绑定死信交换机和routingKey
    channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "DeadKey");
    // 正常队列绑定死信队列信息
    Map params = new HashMap<>();
    // 正常队列设置死信交换机 参数 key 是固定值
    params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
    // 正常队列设置死信 routing-key 参数 key 是固定值
    params.put("x-dead-letter-routing-key", "DeadKey");
    // 声明普通队列，使用参数关联死信队列
    channel.queueDeclare(NORMAL_QUEUE, false, false, false, params);
    // 普通队列绑定普通交换机和routingKey
    channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "InQueue");

    channel.basicConsume(
        NORMAL_QUEUE,
        true,
        (tag, msg) -> {
          String message = new String(msg.getBody(), StandardCharsets.UTF_8);
          System.out.println("Consumer01 接收到消息" + message);
        },
        consumerTag -> {
        }
    );
  }
}

消费者 C2 代码(以上步骤完成后启动 C2 消费者它消费死信队列里面的消息)

package com.zjw.rabbitmq.deadqueue.ttl;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.zjw.rabbitmq.util.RabbitMQUtil;
import java.nio.charset.StandardCharsets;


public class Consumer02 {
  private static final String DEAD_EXCHANGE = "dead_exchange";
  private static final String DEAD_QUEUE = "dead-queue";

  public static void main(String[] argv) throws Exception {
    Channel channel = RabbitMQUtil.getChannel();

    channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
    channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
    channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "DeadKey");
    channel.basicConsume(DEAD_QUEUE, true, (consumerTag, delivery) -> {
      String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
      System.out.println("Consumer02 接收死信队列的消息" + message);
    }, consumerTag -> {
    });
  }
}

6.3.2 队列达到最大长度

生产者一次生产超过队列长度的数据

package com.zjw.rabbitmq.deadqueue.capacity;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.zjw.rabbitmq.util.RabbitMQUtil;
import com.zjw.rabbitmq.util.SleepUtil;


public class Producer {

  private static final String NORMAL_EXCHANGE = "normal_exchange";

  public static void main(String[] argv) throws Exception {
    try (Channel channel = RabbitMQUtil.getChannel()) {
      channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
      //设置消息的 TTL 时间 10s
      BasicProperties properties = new BasicProperties().builder().expiration("60000").build();

      for (int i = 0; i < 10; i++) {
        String message = "hello-" + i;
        channel.basicPublish(NORMAL_EXCHANGE, "InQueue", properties, message.getBytes());
        System.out.println("生产者发送消息:" + message);
        SleepUtil.sleep(3);
      }
    }
  }
}

消费者1定义了队列、交换机之间的绑定关系

package com.zjw.rabbitmq.deadqueue.capacity;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.zjw.rabbitmq.util.RabbitMQUtil;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;


public class Consumer01 {
  // 普通交换机名称
  private static final String NORMAL_EXCHANGE = "normal_exchange";
  // 死信交换机名称
  private static final String DEAD_EXCHANGE = "dead_exchange";
  // 普通队列
  private static final String NORMAL_QUEUE = "normal-queue";
  // 死信队列
  private static final String DEAD_QUEUE = "dead-queue";

  public static void main(String[] argv) throws Exception {
    Channel channel = RabbitMQUtil.getChannel();
    // 声明死信和普通交换机类型为direct
    channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
    channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
    // 声明死信队列
    channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
    // 死信队列绑定死信交换机和routingKey
    channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "DeadKey");
    // 正常队列绑定死信队列信息
    Map params = new HashMap<>();
    // 正常队列设置死信交换机 参数 key 是固定值
    params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
    // 正常队列设置死信 routing-key 参数 key 是固定值
    params.put("x-dead-letter-routing-key", "DeadKey");
    // 设置普通队列长度
    params.put("x-max-length", 5);
    // 声明普通队列，使用参数关联死信队列
    channel.queueDeclare(NORMAL_QUEUE, false, false, false, params);
    // 普通队列绑定普通交换机和routingKey
    channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "InQueue");

    channel.basicConsume(
        NORMAL_QUEUE,
        true,
        (tag, msg) -> {
          String message = new String(msg.getBody(), StandardCharsets.UTF_8);
          System.out.println("Consumer01 接收到消息" + message);
        },
        consumerTag -> {
        }
    );
  }
}

消费者2监控并消费死信队列

package com.zjw.rabbitmq.deadqueue.capacity;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.zjw.rabbitmq.util.RabbitMQUtil;
import java.nio.charset.StandardCharsets;


public class Consumer02 {
  private static final String DEAD_EXCHANGE = "dead_exchange";
  private static final String DEAD_QUEUE = "dead-queue";

  public static void main(String[] argv) throws Exception {
    Channel channel = RabbitMQUtil.getChannel();

    channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
    channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
    channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "DeadKey");
    channel.basicConsume(DEAD_QUEUE, true, (consumerTag, delivery) -> {
      String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
      System.out.println("Consumer02 接收死信队列的消息" + message);
    }, consumerTag -> {
    });
  }
}

6.3.3 消息被拒

Producer生产多条数据

package com.zjw.rabbitmq.deadqueue.reject;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.zjw.rabbitmq.util.RabbitMQUtil;
import com.zjw.rabbitmq.util.SleepUtil;


public class Producer {

  private static final String NORMAL_EXCHANGE = "normal_exchange";

  public static void main(String[] argv) throws Exception {
    try (Channel channel = RabbitMQUtil.getChannel()) {
      channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);

      for (int i = 0; i < 10; i++) {
        String message = "hello-" + i;
        channel.basicPublish(NORMAL_EXCHANGE, "InQueue", null, message.getBytes());
        System.out.println("生产者发送消息:" + message);
        SleepUtil.sleep(3);
      }
    }
  }
}

C1 消费者代码接收消息，其中有消息被拒绝

package com.zjw.rabbitmq.deadqueue.reject;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.zjw.rabbitmq.util.RabbitMQUtil;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;


public class Consumer01 {

  // 普通交换机名称
  private static final String NORMAL_EXCHANGE = "normal_exchange";
  // 死信交换机名称
  private static final String DEAD_EXCHANGE = "dead_exchange";
  // 普通队列
  private static final String NORMAL_QUEUE = "normal-queue";
  // 死信队列
  private static final String DEAD_QUEUE = "dead-queue";

  public static void main(String[] argv) throws Exception {
    Channel channel = RabbitMQUtil.getChannel();
    // 声明死信和普通交换机类型为direct
    channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
    channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
    // 声明死信队列
    channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
    // 死信队列绑定死信交换机和routingKey
    channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "DeadKey");
    // 正常队列绑定死信队列信息
    Map params = new HashMap<>();
    // 正常队列设置死信交换机 参数 key 是固定值
    params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
    // 正常队列设置死信 routing-key 参数 key 是固定值
    params.put("x-dead-letter-routing-key", "DeadKey");
    // 设置普通队列长度
    params.put("x-max-length", 8);
    // 声明普通队列，使用参数关联死信队列
    channel.queueDeclare(NORMAL_QUEUE, false, false, false, params);
    // 普通队列绑定普通交换机和routingKey
    channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "InQueue");
    // 注意关闭自动确认
    channel.basicConsume(NORMAL_QUEUE, false, (tag, msg) -> {
      String message = new String(msg.getBody(), StandardCharsets.UTF_8);
      if (message.equals("hello-4")) {
        System.out.println("Consumer01 接收到消息[" + message + "]并拒绝签收该消息");
        //requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中
        channel.basicReject(msg.getEnvelope().getDeliveryTag(), false);
      } else {
        System.out.println("Consumer01 接收到消息[" + message + "]");
        channel.basicAck(msg.getEnvelope().getDeliveryTag(), false);
      }
    }, consumerTag -> {
    });
  }
}

C2 消费者消费死信队列

package com.zjw.rabbitmq.deadqueue.reject;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.zjw.rabbitmq.util.RabbitMQUtil;
import java.nio.charset.StandardCharsets;


public class Consumer02 {
  private static final String DEAD_EXCHANGE = "dead_exchange";
  private static final String DEAD_QUEUE = "dead-queue";

  public static void main(String[] argv) throws Exception {
    Channel channel = RabbitMQUtil.getChannel();

    channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
    channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
    channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "DeadKey");
    channel.basicConsume(DEAD_QUEUE, true, (consumerTag, delivery) -> {
      String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
      System.out.println("Consumer02 接收死信队列的消息: " + message);
    }, consumerTag -> {
    });
  }
}

注：原始资料来源于尚硅谷，笔者学习过程中做了整理，加入了自己的实验代码

消息队列 — RabbitMQ06

Java相关栏目本月热门文章