发布确认策略

开启发布确认的方法

发布确认默认是没有开启的，如果要开启需要调用方法 confirmSelect，每当你要想使用发布确认，都需要在 channel 上调用该方法

Channel channel = connection.createChannel();
//开启发布确认
channel.confirmSelect();

单个发布确认

这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布，waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。

public class ConfirmMessage {
    public static void main(String[] args) throws Exception {
        //通过类来调用静态方法
        ConfirmMessage.publishMessageIndividually();
    }
    //批量发消息的个数
    public static final int MESSAGE_COUNT = 1000;
    public static void publishMessageIndividually() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName， false， false， false， null);
            //开启发布确认
            channel.confirmSelect();
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "";
                channel.basicPublish(""， queueName， null， message.getBytes());
                //服务端返回 false 或超时时间内未返回，生产者可以消息重发
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println("消息发送成功");
                }
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息，耗时" + (end - begin) + "ms");
        }
    }
}

批量确认发布

上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。

public class ConfirmMessage {
    public static void main(String[] args) throws Exception {
          //通过类来调用静态方法
                ConfirmMessage.publishMessageBatch();
    }
    public static void publishMessageBatch() throws Exception {
        //从工具类中获取信道
        try (Channel channel = RabbitMqUtils.getChannel()) {
            //定义队列
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName， false， false， false， null);
            //开启发布确认
            channel.confirmSelect();
            //批量确认消息大小
            int batchSize = 100;
            //未确认消息个数
            int outstandingMessageCount = 0;
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "";
                channel.basicPublish(""， queueName， null， message.getBytes());
                outstandingMessageCount++;
                //当发送的消息数等于批量确认消息的个数，进行确认，发送消息数归零
                if (outstandingMessageCount == batchSize) {
                    channel.waitForConfirms();
                    outstandingMessageCount = 0;
                }
            }
            //为了确保还有剩余没有确认消息 再次确认
            if (outstandingMessageCount > 0) {
                channel.waitForConfirms();
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息，耗时" + (end - begin) + "ms");
        }
    }
}

异步确认发布

异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功，下面就让我们来详细讲解异步确认是怎么实现的。

public class ConfirmMessage {
    public static void main(String[] args) throws Exception {
          //通过类来调用静态方法
                ConfirmMessage.publishMessageAsync();
    }
    public static void publishMessageAsync() throws Exception {
        //从工具类中获取信道
        try (Channel channel = RabbitMqUtils.getChannel()) {
            //定义队列
            String queueName = UUID.randomUUID().toString();
            //队列声明
            channel.queueDeclare(queueName， false， false， false， null);
            //开启发布确认
            channel.confirmSelect();
            /**
             * 线程安全有序的一个哈希表，适用于高并发的情况
             * 1.轻松的将序号与消息进行关联
             * 2.轻松批量删除条目，只要给到序列号
             * 3.支持并发访问
             */
            ConcurrentSkipListMap<Long， String> outstandingConfirms = new ConcurrentSkipListMap<>();
            /**
             * 确认收到消息的一个回调
             * 1.消息序列号
             * 2.true 可以确认小于等于当前序列号的消息
             * false 确认当前序列号消息
             */
            ConfirmCallback ackCallback = (sequenceNumber， multiple) -> {
                //判断是否为批量消息确认
                if (multiple) {
                    //返回的是小于等于当前序列号的未确认消息 是一个 map
                    ConcurrentNavigableMap<Long， String> confirmed = outstandingConfirms.headMap(sequenceNumber， true);
                    //清除该部分未确认消息（批量），代表它们被确认了
                    confirmed.clear();
                } else {
                    //只清除当前序列号的消息，代表它被确认了
                    outstandingConfirms.remove(sequenceNumber);
                }
            };
            ConfirmCallback nackCallback = (sequenceNumber， multiple) -> {
                String message = outstandingConfirms.get(sequenceNumber);
                System.out.println("发布的消息" + message + "未被确认，序列号" + sequenceNumber);
            };
            /**
             * 添加一个异步确认的监听器
             * 1.确认收到消息的回调
             * 2.未收到消息的回调
             */
            channel.addConfirmListener(ackCallback， nackCallback);
            //发送消息前的时间
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = "消息" + i;
                /**
                 * channel.getNextPublishSeqNo()获取下一个消息的序列号
                 * 通过序列号与消息体进行一个关联
                 * 全部都是未确认的消息体
                 */
                outstandingConfirms.put(channel.getNextPublishSeqNo()， message);
                channel.basicPublish(""， queueName， null， message.getBytes());
            }
            //发送消息后的时间
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息，耗时" + (end - begin) + "ms");
        }
    }
}

如何处理异步未确认消息

最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用 ConcurrentLinkedQueue 这个队列在 confirm callbacks 与发布线程之间进行消息的传递。

以上 3 种发布确认速度对比

单独发布消息

同步等待确认，简单，但吞吐量非常有限。

批量发布消息

批量同步等待确认，简单，合理的吞吐量，一旦出现问题但很难推断出是那条消息出现了问题。

异步处理:

最佳性能和资源使用，在出现错误的情况下可以很好地控制，但是实现起来较难

交换机

在上一节中，我们创建了一个工作队列。我们假设的是工作队列背后，每个任务都恰好交付给一个消费者(工作进程)。在这一部分中，我们将做一些完全不同的事情-我们将消息传达给多个消费者。这种模式称为 ”发布/订阅”.为了说明这种模式，我们将构建一个简单的日志系统。它将由两个程序组成:第一个程序将发出日志消息，第二个程序是消费者。其中我们会启动两个消费者，其中一个消费者接收到消息后把日志存储在磁盘，另外一个消费者接收到消息后把消息打印在屏幕上，事实上第一个程序发出的日志消息将广播给所有消费者。

Exchanges

Exchanges概念

RabbitMQ 消息传递模型的核心思想是: 生产者生产的消息从不会直接发送到队列。实际上，通常生产者甚至都不知道这些消息传递传递到了哪些队列中。

相反，生产者只能将消息发送到交换机(exchange)，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列还是说把他们到许多队列中还是说应该丢弃它们。这就的由交换机的类型来决定。

Exchanges的类型

总共有以下类型:

直接(direct)，主题(topic) ，标题(headers) ，扇出(fanout)

无名exchange

在本教程的前面部分我们对 exchange 一无所知，但仍然能够将消息发送到队列。之前能实现的原因是因为我们使用的是默认交换，我们通过空字符串(“”)进行标识。

channel.basicPublish(""， "hello"， null， message.getBytes());

第一个参数是交换机的名称。空字符串表示默认或无名称交换机:消息能路由发送到队列中其实是由 routingKey(bindingkey)绑定 key 指定的，如果它存在的话。

临时队列

之前的章节我们使用的是具有特定名称的队列(还记得 hello 和 ack_queue 吗？)。队列的名称我们来说至关重要-我们需要指定我们的消费者去消费哪个队列的消息。

每当我们连接到 Rabbit 时，我们都需要一个全新的空队列，为此我们可以创建一个具有随机名称的队列，或者能让服务器为我们选择一个随机队列名称那就更好了。其次一旦我们断开了消费者的连接，队列将被自动删除。

创建临时队列的方式如下:

String queueName = channel.queueDeclare().getQueue();

创建出来长这样：

绑定（bindings）

什么是 bingding 呢，binding 其实是 exchange 和 queue 之间的桥梁，它告诉我们 exchange 和那个队列进行了绑定关系。比如说下面这张图告诉我们的就是 X 与 Q1 和 Q2 进行了绑定

Fanout

Fanout介绍

Fanout 这种类型非常简单。正如从名称中猜到的那样，它是将接收到的所有消息广播到它知道的所有队列中。系统中默认有些 exchange 类型。

Fanout实战

Logs 和临时队列的绑定关系如下图

ReceiveLogs01 将接收到的消息打印在控制台

public class ReceiveLogs01 {
    //交换机的名称
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //声明一个交换机
        channel.exchangeDeclare(EXCHANGE_NAME， "fanout");
        /**
         * 生成一个临时的队列 队列的名称是随机的
         * 当消费者断开和该队列的连接时 队列自动删除
         */
        String queueName = channel.queueDeclare().getQueue();
        //把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串
        channel.queueBind(queueName， EXCHANGE_NAME， "");
        System.out.println("等待接收消息，把接收到的消息打印在屏幕........... ");
        DeliverCallback deliverCallback = (consumerTag， delivery) -> {
            String message = new String(delivery.getBody()， "UTF-8");
            System.out.println("控制台打印接收到的消息" + message);
        };
        //接收消息
        channel.basicConsume(queueName， true， deliverCallback， consumerTag -> {
        });
    }
}

ReceiveLogs02 将接收到的消息存储在磁盘

public class ReceiveLogs02 {
    //交换机的名称
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //声明一个交换机
        channel.exchangeDeclare(EXCHANGE_NAME， "fanout");
        /**
         * 生成一个临时的队列 队列的名称是随机的
         * 当消费者断开和该队列的连接时 队列自动删除
         */
        String queueName = channel.queueDeclare().getQueue();
        //把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串
        channel.queueBind(queueName， EXCHANGE_NAME， "");
        System.out.println("等待接收消息，把接收到的消息写到文件........... ");
        DeliverCallback deliverCallback = (consumerTag， delivery) ->
        {String message = new String(delivery.getBody()， "UTF-8");
            File file = new File("D:\\rabbitmq_info.txt");
            FileUtils.writeStringToFile(file，message，"UTF-8");
            System.out.println("数据写入文件成功");
        };
        //接收消息
        channel.basicConsume(queueName， true， deliverCallback， consumerTag -> {
        });
    }
}

EmitLog 发送消息给两个消费者接收

public class EmitLog {
    //交换机名称
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            /**
             * 声明一个 exchange
             * 1.exchange 的名称
             * 2.exchange 的类型
             */
            channel.exchangeDeclare(EXCHANGE_NAME， "fanout");
            Scanner sc = new Scanner(System.in);
            System.out.println("请输入信息");
            while (sc.hasNext()) {
                String message = sc.nextLine();
                channel.basicPublish(EXCHANGE_NAME， ""， null， message.getBytes("UTF-8"));
                System.out.println("生产者发出消息" + message);
            }
        }
    }
}

Direct exchange

回顾

在上一节中，我们构建了一个简单的日志记录系统。我们能够向许多接收者广播日志消息。在本节我们将向其中添加一些特别的功能-比方说我们只让某个消费者订阅发布的部分消息。例如我们只把严重错误消息定向存储到日志文件(以节省磁盘空间)，同时仍然能够在控制台上打印所有日志消息。

我们再次来回顾一下什么是 bindings，绑定是交换机和队列之间的桥梁关系。也可以这么理解:队列只对它绑定的交换机的消息感兴趣。绑定用参数:routingKey 来表示也可称该参数为 binding key，创建绑定我们用代码:channel.queueBind(queueName， EXCHANGE_NAME， "routingKey");绑定之后的意义由其交换类型决定。

Direct exchange 介绍

上一节中的我们的日志系统将所有消息广播给所有消费者，对此我们想做一些改变，例如我们希望将日志消息写入磁盘的程序仅接收严重错误(errros)，而不存储哪些警告(warning)或信息(info)日志消息避免浪费磁盘空间。Fanout 这种交换类型并不能给我们带来很大的灵活性-它只能进行无意识的广播，在这里我们将使用 direct 这种类型来进行替换，这种类型的工作方式是，消息只去到它绑定的routingKey 队列中去。

在上面这张图中，我们可以看到X绑定了两个队列，绑定类型是direct。队列Q1绑定键为orange，队列Q2绑定键有两个:一个绑定键为black，另一个绑定键为green。

在这种绑定情况下，生产者发布消息到exchange上，绑定键为orange的消息会被发布到队列Q1。绑定键为blackgreen和的消息会被发布到队列Q2，其他消息类型的消息将被丢弃。

多重绑定

当然如果exchange的绑定类型是direct，但是它绑定的多个队列的key如果都相同，在这种情况下虽然绑定类型是direct但是它表现的就和fanout有点类似了，就跟广播差不多，如上图所示。

实战

Exchange：direct_logs

public class ReceiveLogsDirect01 {
    //交换机
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args) throws Exception {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        //声明一个交换机，交换机名称，枚举类型
        channel.exchangeDeclare(EXCHANGE_NAME， BuiltinExchangeType.DIRECT);
        //队列名称
        String queueName = "disk";
        //声明一个队列
        channel.queueDeclare(queueName， false， false， false， null);
        //绑定队列和交换机
        channel.queueBind(queueName， EXCHANGE_NAME， "error");
        System.out.println("等待接收消息........... ");
        DeliverCallback deliverCallback = (consumerTag， delivery) -> {
            String message = new String(delivery.getBody()， "UTF-8");
            message = "接收绑定键:" + delivery.getEnvelope().getRoutingKey() + "，消息:" + message;
            File file = new File("D:\\rabbitmq_info.txt");
            FileUtils.writeStringToFile(file， message， "UTF-8");
            System.out.println("错误日志已经接收");
        };
        //接收消息
        channel.basicConsume(queueName， true， deliverCallback， consumerTag -> {
        });
    }
}
public class ReceiveLogsDirect02 {
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME， BuiltinExchangeType.DIRECT);
        String queueName = "console";
        channel.queueDeclare(queueName， false， false， false， null);
        channel.queueBind(queueName， EXCHANGE_NAME， "info");
        channel.queueBind(queueName， EXCHANGE_NAME， "warning");
        System.out.println("等待接收消息........... ");
        DeliverCallback deliverCallback = (consumerTag， delivery) -> {
            String message = new String(delivery.getBody()， "UTF-8");
            System.out.println(" 接收绑定键 :" + delivery.getEnvelope().getRoutingKey() + "， 消 息:" + message);
        };
        channel.basicConsume(queueName， true， deliverCallback， consumerTag -> {
        });
    }
}
public class EmitLogDirect {
    //交换机名称
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args) throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME， BuiltinExchangeType.DIRECT);
            //创建多个 bindingKey
            Map<String， String> bindingKeyMap = new HashMap<>();
            bindingKeyMap.put("info"， "普通 info 信息");
            bindingKeyMap.put("warning"， "警告 warning 信息");
            bindingKeyMap.put("error"， "错误 error 信息");
            //debug 没有消费这接收这个消息 所以就丢失了
            bindingKeyMap.put("debug"， "调试 debug 信息");
            for (Map.Entry<String， String> bindingKeyEntry : bindingKeyMap.entrySet()) {
                String bindingKey = bindingKeyEntry.getKey();
                String message = bindingKeyEntry.getValue();
                channel.basicPublish(EXCHANGE_NAME， bindingKey， null， message.getBytes("UTF-8"));
                System.out.println("生产者发出消息:" + message);
            }
        }
    }
}

Topics

之前类型的问题

在上一个小节中，我们改进了日志记录系统。我们没有使用只能进行随意广播的 fanout 交换机，而是使用了 direct 交换机，从而有能实现有选择性地接收日志。

尽管使用direct 交换机改进了我们的系统，但是它仍然存在局限性-比方说我们想接收的日志类型有info.base 和 info.advantage，某个队列只想 info.base 的消息，那这个时候direct 就办不到了。这个时候就只能使用 topic 类型

Topic的要求

发送到类型是 topic 交换机的消息的 routing_key 不能随意写，必须满足一定的要求，它必须是一个单词列表，以点号分隔开。这些单词可以是任意单词，比如:"stock.usd.nyse"， "nyse.vmw"，"quick.orange.rabbit".这种类型的。当然这个单词列表最多不能超过 255 个字节。

在这个规则列表中，其中有两个替换符是大家需要注意的

*(星号)可以代替一个单词

#(井号)可以替代零个或多个单词

Topic匹配案例

下图绑定关系如下

Q1-->绑定的是

中间带 orange 带 3 个单词的字符串(.orange.)

Q2-->绑定的是

最后一个单词是 rabbit 的 3 个单词(..rabbit)

第一个单词是 lazy 的多个单词(lazy.#)

上图是一个队列绑定关系图，我们来看看他们之间数据接收情况是怎么样的

quick.orange.rabbit 被队列 Q1Q2 接收到

lazy.orange.elephant 被队列 Q1Q2 接收到

quick.orange.fox 被队列 Q1 接收到

lazy.brown.fox 被队列 Q2 接收到

lazy.pink.rabbit 虽然满足两个绑定但只被队列 Q2 接收一次

quick.brown.fox 不匹配任何绑定不会被任何队列接收到会被丢弃

quick.orange.male.rabbit 是四个单词不匹配任何绑定会被丢弃

lazy.orange.male.rabbit 是四个单词但匹配 Q2

当队列绑定关系是下列这种情况时需要引起注意

当一个队列绑定键是#，那么这个队列将接收所有数据，就有点像 fanout 了
如果队列绑定键当中没有#和*出现，那么该队列绑定类型就是direct了

实战

public class EmitLogTopic {
    private static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME， "topic");
            /**
             * Q1-->绑定的是
             * 中间带 orange 带 3 个单词的字符串(*.orange.*)
             * Q2-->绑定的是
             * 最后一个单词是 rabbit 的 3 个单词(*.*.rabbit)
             * 第一个单词是 lazy 的多个单词(lazy.#)
             *
             */
            Map<String， String> bindingKeyMap = new HashMap<>();
            bindingKeyMap.put("quick.orange.rabbit"， "被队列 Q1Q2 接收到");
            bindingKeyMap.put("lazy.orange.elephant"， "被队列 Q1Q2 接收到");
            bindingKeyMap.put("quick.orange.fox"， "被队列 Q1 接收到");
            bindingKeyMap.put("lazy.brown.fox"， "被队列 Q2 接收到");
            bindingKeyMap.put("lazy.pink.rabbit"， "虽然满足两个绑定但只被队列 Q2 接收一次");
            bindingKeyMap.put("quick.brown.fox"， "不匹配任何绑定不会被任何队列接收到会被丢弃");
            bindingKeyMap.put("quick.orange.male.rabbit"， "是四个单词不匹配任何绑定会被丢弃");
            bindingKeyMap.put("lazy.orange.male.rabbit"， "是四个单词但匹配 Q2");
            for (Map.Entry<String， String> bindingKeyEntry : bindingKeyMap.entrySet()) {
                String bindingKey = bindingKeyEntry.getKey();
                String message = bindingKeyEntry.getValue();
                channel.basicPublish(EXCHANGE_NAME， bindingKey， null， message.getBytes("UTF-8"));
                System.out.println("生产者发出消息" + message);
            }
        }
    }
public class ReceiveLogsTopic01 {
    private static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME， "topic");
        //声明 Q1 队列与绑定关系
        String queueName = "Q1";
        channel.queueDeclare(queueName， false， false， false， null);
        channel.queueBind(queueName， EXCHANGE_NAME， "*.orange.*");
        System.out.println("等待接收消息........... ");
        DeliverCallback deliverCallback = (consumerTag， delivery) -> {
            String message = new String(delivery.getBody()， "UTF-8");
            System.out.println(" 接 收 队 列 :" + queueName + " 绑 定 键:" + delivery.getEnvelope().getRoutingKey() + "，消息:" + message);
        };
        channel.basicConsume(queueName， true， deliverCallback， consumerTag -> {
        });
    }
}
public class ReceiveLogsTopic02 {
    private static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME， "topic");
        //声明 Q2 队列与绑定关系
        String queueName = "Q2";
        channel.queueDeclare(queueName， false， false， false， null);
        channel.queueBind(queueName， EXCHANGE_NAME， "*.*.rabbit");
        channel.queueBind(queueName， EXCHANGE_NAME， "lazy.#");
        System.out.println("等待接收消息........... ");
        DeliverCallback deliverCallback = (consumerTag， delivery) -> {
            String message = new String(delivery.getBody()， "UTF-8");
            System.out.println(" 接 收 队 列 :" + queueName + " 绑 定 键:" + delivery.getEnvelope().getRoutingKey() + "，消息:" + message);
        };
        channel.basicConsume(queueName， true， deliverCallback， consumerTag -> {
        });
    }
}

死信队列

死信的概念

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

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

死信的来源

消息 TTL 过期

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

死信实战

代码架构图

消息TTL过期

生产者代码

public class Producer {
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    public static void main(String[] args) throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            //设置消息的 TTL 时间
            AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder()
                .expiration("10000")
                .build();
            //该信息是用作演示队列个数限制
            for (int i = 1; i < 11; i++) {
                String message = "info" + i;
                channel.basicPublish(NORMAL_EXCHANGE， "zhangsan"， null， message.getBytes());
                System.out.println("生产者发送消息:" + message);
            }
        }
    }
}
消费者C1代码（启动只后关闭消费者，模拟其接收不到消息）
public class Consumer01 {
    //普通交换机名称
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";
    //普通队列名称
    private static final String normalQueue = "normal-queue";
    //死信队列名称
    private static final String String deadQueue = "dead-queue";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //声明死信和普通交换机 类型为 direct
        channel.exchangeDeclare(NORMAL_EXCHANGE， BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE， BuiltinExchangeType.DIRECT);
        //声明死信队列
        channel.queueDeclare(deadQueue， false， false， false， null);
        //死信队列绑定死信交换机与 routingkey
        channel.queueBind(deadQueue， DEAD_EXCHANGE， "lisi");
        //正常队列绑定死信队列信息
        Map<String， Object> params = new HashMap<>();
        //正常队列设置死信交换机 参数 key 是固定值
        params.put("x-dead-letter-exchange"， DEAD_EXCHANGE);
        //正常队列设置死信 routing-key 参数 key 是固定值
        params.put("x-dead-letter-routing-key"， "lisi");
        //声明正常队列
        channel.queueDeclare(normalQueue， false， false， false， params);
        //绑定正常队列
        channel.queueBind(normalQueue， NORMAL_EXCHANGE， "zhangsan");
        System.out.println("等待接收消息........... ");
        DeliverCallback deliverCallback = (consumerTag， delivery) -> {
            String message = new String(delivery.getBody()， "UTF-8");
            System.out.println("Consumer01 接收到消息" + message);
        };
        //接收消息
        channel.basicConsume(normalQueue， true， deliverCallback， consumerTag -> {
        });
    }
}

生产者未发送消息

生产者发送了10条消息，此时正常消息队列有10条未消费消息

时间过去10秒，正常队列里面的消息由于没有被消费，消息进入死信队列

消费者C2代码（完成上述步骤后启动C2消费者，它消费死信队列里面的消息）

public class Consumer02 {
    private static final String DEAD_EXCHANGE = "dead_exchange";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(DEAD_EXCHANGE， BuiltinExchangeType.DIRECT);
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue， false， false， false， null);
        channel.queueBind(deadQueue， DEAD_EXCHANGE， "lisi");
        System.out.println("等待接收死信队列消息........... ");
        DeliverCallback deliverCallback = (consumerTag， delivery) -> {
            String message = new String(delivery.getBody()， "UTF-8");
            System.out.println("Consumer02 接收死信队列的消息" + message);
        };
        channel.basicConsume(deadQueue， true， deliverCallback， consumerTag -> {
        });
    }
}

死信队列里面的消息被C2消费

队列达到最大长度

生产者代码去掉TTL属性

public class Producer {
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    public static void main(String[] args) throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            //该信息是用作演示队列个数限制
            for (int i = 1; i < 11; i++) {
                String message = "info" + i;
                channel.basicPublish(NORMAL_EXCHANGE， "zhangsan"， null， message.getBytes());
                System.out.println("生产者发送消息:" + message);
            }
        }
    }
}

C1消费者修改以下代码（启动之后关闭该消费者，模拟其接收不到消息）

public class Consumer01 {
    //普通交换机名称
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";
    //普通队列名称
    private static final String normalQueue = "normal-queue";
    //死信队列名称
    private static final String String deadQueue = "dead-queue";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //声明死信和普通交换机 类型为 direct
        channel.exchangeDeclare(NORMAL_EXCHANGE， BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE， BuiltinExchangeType.DIRECT);
        //声明死信队列
        channel.queueDeclare(deadQueue， false， false， false， null);
        //死信队列绑定死信交换机与 routingkey
        channel.queueBind(deadQueue， DEAD_EXCHANGE， "lisi");
        //正常队列绑定死信队列信息
        Map<String， Object> params = new HashMap<>();
        //正常队列设置死信交换机 参数 key 是固定值
        params.put("x-dead-letter-exchange"， DEAD_EXCHANGE);
        //正常队列设置死信 routing-key 参数 key 是固定值
        params.put("x-dead-letter-routing-key"， "lisi");
        //设置正常队列的长度限制，长度限制为6，超出的部分进入死信队列，先进入队列的部分
        params.put("x-max-length"， 6);
        //声明正常队列
        channel.queueDeclare(normalQueue， false， false， false， params);
        //绑定正常队列
        channel.queueBind(normalQueue， NORMAL_EXCHANGE， "zhangsan");
        System.out.println("等待接收消息........... ");
        DeliverCallback deliverCallback = (consumerTag， delivery) -> {
            String message = new String(delivery.getBody()， "UTF-8");
            System.out.println("Consumer01 接收到消息" + message);
        };
        //接收消息
        channel.basicConsume(normalQueue， true， deliverCallback， consumerTag -> {
        });
    }
}

消息队列（四）