RabbitMQ笔记

消息队列

1.1 MQ相关概念

1.1.1 什么是MQ

MQ(message queue),从字面意思上看，本质是个队列，FIFO先入先出，只不过队列中存放的内容是
message而已，还是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ是一种非常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用了MQ之后，消息发送上游只需要依赖MQ,不用依赖其他服务。

1.1.2 为什么要用MQ

1. 流量消峰

   比如一个系统1s只能处理一万个下单请求，但在高峰时1s内有2万个下单请求，这时候系统就会崩溃，此时就需要将请求转给MQ，达到消峰的目的。

   缺点：请求到达MQ需要排队，导致用户在下单十几秒之后才收到下单成功的操作。性能差。

   

2. 应用解耦

   以电商应用为例，应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。当转变成基于消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，中间用户感受不到物流系统的故障，提升系统的可用性。

   

3. 异步处理

   B处理完成后向MQ发送消息，MQ又会通知A B的操作已经执行完

   

1.1.3 MQ的分类

1. Active MQ

   优点：单机吞吐量万级，时效性s级，可用性高，基于主从架构实现高可用性，消息可靠性较低的概率丢失数据
   缺点：官方社区现在对ActiveMQ5.x维护越来越少，高吞吐量场景较少使用。
   尚硅谷官网视频：http:/www.gulixueyuan.com/course/322

2. kafka

   大数据的杀手锏，谈到大数据领域内的消息传输，则绕不开Kfka,这款为大数据而生的消息中间件，
   以其百万级TPS的吞吐量名声大噪，迅速成为大数据领域的宠儿，在数据采集、传输、存储的过程中发挥着举足轻重的作用。目前已经被LinkedIn,Uber,Twitter,Netflix等大公司所采纳。

   优点：性能卓越，单机写入TPS约在百万条/秒，最大的优点，就是吞吐量高。时效性s级可用性非常高，kaka是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用，消费者采用Pu方式获取消息，消息有序，通过控制能够保证所有消息被消费且仅被消费一次；有优秀的第三方Kafka Web管理界面Kafka-Manager;在日志领域比较成熟，被多家公司和多个开源项目使用；功能支持：功能较为简单，主要支持简单的MQ功能，在大数据领域的实时计算以及日志采集被大规模使用

   缺点：Kafka单机超过64个队列/分区，Load会发生明显的飙高现象，队列越多，load越高，发送消
   息响应时间变长，使用短轮询方式，实时性取决于轮询间隔时间，消费失败不支持重试，支持消息顺序，但是一台代理宕机后，就会产生消息乱序，社区更新较慢。

   适合产生大量数据的互联网服务的数据收集业务，有日志采集功能就选它。

3. Rocket MQ

   RocketMQ出自阿里巴巴的开源产品，用]ava语言实现，在设计时参考了Kafka,并做出了自己的一些改进。被阿里巴巴广泛应用在订单，交易，充值，流计算，消息推送，日志流式处理，binglog分发等场景。

   优点：单机吞吐量十万级可用性非常高，分布式架构，消息可以做到0丢失MQ功能较为完善，还是分
   布式的，扩展性好，支持10亿级别的消息堆积，不会因为堆积导致性能下降，源码是java我们可以自己阅读源码，定制自己公司的MQ
   缺点：支持的客户端语言不多，目前是java及c++,其中c++不成熟；社区活跃度一般，没有在MQ核心中去实现JMS等接口，有些系统要迁移需要修改大量代码

   天生为金融互联网而生，对于可靠性要求很高的场景。

4. Rabbit MQ

   2007年发布，是一个在AMQP(高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一。
   优点：由于erlang语言的高并发特性，性能较好；吞吐量到万级，MQ功能比较完备，健壮、稳定、易用、跨平台x支持多种语言如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持A]X文档齐全；开源提供的管理界面非常棒，用起来很好用，社区活跃度高；更新频率相当高https://www.rabbitmq.com/news.html
   缺点：商业版需要收费，学习成本较高

适合数据量不是特别大的场景。

1.2 Rabbit MQ

1.2.1 Rabbit MQ的概念

RabbitMQ是一个消息中间件：它接收并转发消息，但不处理。

1.2.2 四大核心概念

• 生产者
• 交换机
• 队列
• 消费者

1.2.3 Rabbit MQ核心部分

1.2.4 各个名词介绍

Broker：接收和分发消息的应用，Rabbit MQ Server就是Message Broker

Connection：消费者/生产者和broker之间的TCP连接

Channel:如果每一次访问RabbitMQ都建立一个Connection,在消息量大的时候建立TCP
Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接，如果应用程序支特多线程，通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id帮助客户端和message broker识别channel,,所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销

Virtual host：出于多租户和安全因素设计的，把AMQP的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost,每个用户在自己的vhost创建exchange/queue等

Exchange：message到达broker的第一站，根据分发规则，匹配查询表中的routing key,分发消息到queue中去。常用的类型有：direct(point-.to-point),topic(publish-subscribe)and fanout(multicast)
Queue：消息最终被送到这里等待consumer取走
Binding：exchange和queue之间的虚拟连接，binding中可以包含routing key,Binding信息被保存到exchange中的查询表中，用于message的分发依据

2. hello world——RabbitMQ初使用

2.1 引入依赖

JAVA
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.18.0</version>
</dependency>
<dependency>
    <groupId>commons-io</groupId>
    <artifactId>commons-io</artifactId>
    <version>2.6</version>
</dependency>

2.2 创建生产者和消费者

生产者：

JAVA
public class Producer1 {
    // 队列名称
    public static final String QUEUE_NAME = "hello";

    // 发消息
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置工厂 IP 连接RabbitMQ的队列
        factory.setHost("localhost");
        // 用户名和密码
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 获取信道
        // 创建连接
        Connection connection = factory.newConnection();
        // 获取信道
        Channel channel = connection.createChannel();
        /**
         * 生成一个队列
         * 1.队列名称
         * 2.队列中消息是否持久化（存在磁盘中） 默认情况下消息存在内容中
         * 3.该队列是否只供一个消费者消费 （是否进行消息共享）true：可以多个消费者消费  false：只能一个消费者消费
         * 4.是否自动删除 最后一个消费者端开连接以后该队一句是否自动删除true自动删除false不自动删除
         * 5.其它参数
         */

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 发消息
        String message = "hello world";
        /**
         * 发送一个消费
         * 1.发送到哪个交换机
         * 2.路由的Key值是哪个 本次是队列的名称
         * 3.其它参数信息
         * 4.发送消息的本体
         */
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("消息发送完毕");
    }
}

消费者

JAVA
public class Consumer1 {
    // 队列的名称
    public static final String QUEUE_NAME = "hello";
    // 接收消息
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        // 创建信道 
        Channel channel = connection.createChannel();

        // 声明 接收到了消息
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println(new String(message.getBody()));
        };

        // 声明 取消接收消息时的回调
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消息消费被中断");
        };
        /**
         * 消费者消费信息
         * 1.消费哪个队列
         * 2.消费成功后是否要应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费的回调
         */

         channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}

3. Work Queues（工作队列）

3.1 轮询分发消息

可以创建两个工作线程/消费者接收消息，创建一个什么生产者不断发送消息，这时候会发现有一个线程成功接收第一次发送消息，则不会接收第二次发送的消息，而第二次发送的消息则是被另一个线程接收了。即两个工作线程不能同时接收同一个消息，消息的接收是一个轮询机制。

3.2 消息应答

3.2.1概念

为了保证消息在发送过程中不丢失，RabbitMQ引入消息应答机制，消息应答就是：消费者在接收到消息并且处理该消息之后，告诉RabbitMQ它己经处理了，RabbitMQ可以把该消息删除了。

3.2.2 自动应答

这种模式需要在高吞吐量和数据传输安全性方面做权衡。仅适用于在消费者可以高效并以某种速率能够处理这些消息的情况下使用。

一旦消息被接收了就会应答，如果消息处理失败了，还是会应答，这样会导致消息的丢失。

3.2.3 消息应答的方式

• Channel.basicAck(用于肯定确认) RabbitMQ已知道该消息并且成功的处理消息，可以将其丢弃了 开启自动应答
• Channel.basicNack(用于否定确认) 关闭自动应答
• Channel.basicReject(用于否定确认)，与Channel.basicNack相比少一个参数。不处理该消息了直接拒绝，可以将其丢弃了

3.2.4 Multiple的解释

手动应答好处是可以批量应答且减少网络拥堵

3.2.5 消息自动重新入队

如果消费者因为某些原因失去了连接，导致消息未发送ACK确认，MQ则知道消息未完全处理，并将此消息重新入队。如果其他其他消费者可以处理，它将会很快安排其他消费者进行处理。

3.2.6 MQ开启手动应答

MQ开启手动应答消息是不会丢失的，如果有一个工作线程挂掉了，本该被C2处理的消息会被转发到C1线程进行处理

3.6.7 手动应答实战

生产者：

JAVA
/**
 * 消息在手动应答时是不丢，放回队列中重新消费
 */
public class Task2 {
    public static final String task_queue_name = "ack_queue";

    public static void main(String[] args) throws IOException {
        Channel channel = (Channel) RabbitMQUtils.getChannel();
        // 声明队列
        boolean durable = true;
        channel.queueDeclare(task_queue_name, durable, false, false, null);
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String message = scanner.next();
            // 设置生产者发送消息为持久化消息（要求保存在磁盘中） 不设置则保存在内存中
            channel.basicPublish("", task_queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
            System.out.println("生产者发送消息：" + message);
        }
    }
}

消费者1：

JAVA
/**
 * 消息在手动应答时是不丢失的，放回队列中重新消费
 */
public class Work03 {
    // 队列名称
    public static final String task_queue_name = "ack_queue";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        System.out.println("C1等待接收消息处理时间较短");
        DeliverCallback deliverCallback = (consumerTag,message) -> {
            SleepUtils.sleep(1);
            System.out.println("接收到的消息" + new String(message.getBody()));
            // 手动应答
            /**
             * 1.消息的标记 tag
             * 2.是否批量应答 false不批量应答信道中的消息 反之批量
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
        };
        CancelCallback cancelCallback = consumerTag -> {

        };
        // 设置不公平分发
        int prefetchCount = 2;
        channel.basicQos(prefetchCount);
        // 采用手动应答
        boolean autoAck = false;
        channel.basicConsume(task_queue_name, autoAck, deliverCallback, cancelCallback);
    }
}

消费者2：

JAVA
public class Work04 {
        // 队列名称
        public static final String task_queue_name = "ack_queue";
        public static void main(String[] args) throws IOException {
            Channel channel = RabbitMQUtils.getChannel();
            System.out.println("C2等待接收消息处理时间较长");
            DeliverCallback deliverCallback = (consumerTag, message) -> {
                SleepUtils.sleep(30);
                System.out.println("接收到的消息" + new String(message.getBody()));
                // 手动应答
                /**
                 * 1.消息的标记 tag
                 * 2.是否批量应答 false不批量应答信道中的消息 反之批量
                 */
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            };
            CancelCallback cancelCallback = consumerTag -> {
                System.out.println("已取消");
            };
            int prefetchCount = 5;
            channel.basicQos(prefetchCount);
            channel.basicConsume(task_queue_name, false, deliverCallback, cancelCallback);
        }
}

3.3 RabbitMQ持久化

3.3.1 概念

之前我们创建的队列都是非持久化的，rabbitmq如果重启的话，该队列就会被删除掉，如果
要队列实现持久化需要在声明队列的时候把durable参数设置为持久化。

3.3.2 队列如何实现持久化

在声明队列时，将durable改为true

JAVA
boolean durable = true;
channel.queueDeclare(*queue_name*, durable, false, false, null);

3.3.3 消息实现持久化

将消息标记为持久化时并不能完全保证不会丢失消息，因为其在刚准备存入磁盘时，但还没有存储完磁盘中，消息还在缓存的一个间隔点，此时并没有真正存入磁盘，持久性保证并不强。

JAVA
channel.basicPublish("", task_queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));

3.3.4 不公平分发（能者多劳）

RabbitMQ分发消息采用轮训分发，如果有的消费者消费慢，但有的消费快，其会导致快的消费者很闲，所以应该采用不公平分发。

JAVA
channel.basicQos(1); //设置在消费方

3.3.5预取值

预先设置消费者消费的消息数，通过channel信道设置。prefetchCount指的是信道的积压的信息数。

JAVA
int prefetchCount = 2; // 要求 > 1
channel.basicQos(prefetchCount);

4. 发布确认

4.1 发布确认原理

1. 要求队列必须持久化
2. 消息也必须持久化
3. 成功保存在磁盘上后才会进行发布确认（存在磁盘后就会告诉生产者）

4.2 发布确认的策略

4.2.1 单个确认发布

发布一个消息确认一个，性能差，但消息丢失能知道哪个消息丢失了

发布一千个消息用时368ms

PLAINTEXT
public static void publishMessageIndividually() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        // 队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, 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());
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功");
            }
        }
        // 结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布1000个消息单个确认耗时" + (end-begin) +"ms");
    }

4.2.2 批量确认发布

发布完一堆消息后再进行统一确认，如果有消息出问题难以定位出错的是哪一个

发布一千个消息用时48ms

PLAINTEXT
public static void publishMessageBatch() throws IOException, InterruptedException {
    Channel channel = RabbitMQUtils.getChannel();
    // 队列的声明
    String queueName = UUID.randomUUID().toString();
    channel.queueDeclare(queueName, true, false, false, null);
    // 开启确认发布
    channel.confirmSelect();
    // 开启时间
    long begin = System.currentTimeMillis();

    //批量确认消息的大小
    int batchSize = 1000;
    for (int i = 0; i< MESSAGE_COUNT; i++) {
        String message = i+"";
        channel.basicPublish("", queueName, null, message.getBytes());

        if (i%batchSize == 0) {
            // 发布确认
            channel.waitForConfirms();
        }
    }
    // 结束时间
    long end = System.currentTimeMillis();
    System.out.println("发布1000个消息批量确认耗时" + (end-begin) +"ms");
}

4.2.3 异步确认发布

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

PLAINTEXT
// 异步确认发布
    public static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        // 队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启确认发布
        channel.confirmSelect();
        // 开启时间
        long begin = System.currentTimeMillis();

        // 消息确认成功的回调方法
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            System.out.println("确认的消息：" + deliveryTag);
        };
        // 消息确认失败的回调方法
        /**
         * 1.消息的标识 deliveryTag
         * 2.是否批量 multiple
         */
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("未确认的消息：" + deliveryTag);
        };

        // 准备消息的监听器 监听哪些消息成功了 哪些些消息失败了
        /**
         * 1.监听哪些消息成功了
         * 2.监听那些消息失败了
         */
        channel.addConfirmListener(ackCallback, nackCallback);  // 监听器有两种，一种只监听成功的，一种成功的和失败的都监听，这里的话是后者

        // 批量确认发布
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i+"";
            channel.basicPublish("", queueName, null, message.getBytes());
            // 发布确认 异步

        }

        // 结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布1000个消息异步确认发布耗时" + (end-begin) +"ms");
    }

如何处理异步未确认的消息

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

PLAINTEXT
// 异步确认发布
public static void publishMessageAsync() throws Exception{
    Channel channel = RabbitMQUtils.getChannel();
    // 队列的声明
    String queueName = UUID.randomUUID().toString();
    channel.queueDeclare(queueName, true, false, false, null);
    // 开启确认发布
    channel.confirmSelect();
    // 开启时间
    long begin = System.currentTimeMillis();

    /**
     * 线程安全有序的一个哈希表 适用于高并发的情况下
     * 1.轻松的将序号与消息进行关联
     * 2.轻松批量删除条目，只要给到序号
     * 3.支持高并发（多线程）
     */
    ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

    // 消息确认成功的回调方法
    ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
        // 如果是批量发布消息，则批量删除
        if (multiple) {
            // 2.删除掉已经确认的消息，剩下就是未确认的消息
            ConcurrentNavigableMap<Long, String> confirmedMap =
                    outstandingConfirms.headMap(deliveryTag);
            confirmedMap.clear();
        } else {
            // 单个确认
            outstandingConfirms.remove(deliveryTag);
        }

        System.out.println("确认的消息是：" + deliveryTag);
    };
    // 消息确认失败的回调方法
    /**
     * 1.消息的标识 deliveryTag
     * 2.是否批量 multiple
     */
    ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
        String message = outstandingConfirms.get(deliveryTag);
        // 3.打印未确认的消息有哪些
        System.out.println("未确认的消息是：" + message + "未确认消息的标记是：" + deliveryTag);
    };
    
    // 准备消息的监听器 监听哪些消息成功了 哪些些消息失败了
    /**
     * 1.监听哪些消息成功了
     * 2.监听那些消息失败了
     */
    channel.addConfirmListener(ackCallback, nackCallback);  // 监听器有两种，一种只监听成功的，一种成功的和失败的都监听，这里的话是后者

    // 批量确认发布
    for (int i = 0; i < MESSAGE_COUNT; i++) {
        String message = i+"";
        channel.basicPublish("", queueName, null, message.getBytes());
        // 1. 此处记录下所有要发送消息，消息的总和
        outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
        // 发布确认 异步

    }
    // 结束时间
    long end = System.currentTimeMillis();
    System.out.println("发布1000个消息异步确认发布耗时" + (end-begin) +"ms");
}

4.2.4 三种发布的对比

单独发布消息
同步等待确认，简单，但吞吐量非常有限。
批量发布消息
批量同步等待确认，简单，合理的吞吐量，一旦出现问题但很难推断出是那条
消息出现了问题。
异步处理：
最佳性能和资源使用，在出现错误的情况下可以很好地控制，但是实现起来稍微难些

4.2.5 发布确认实战

ConfirmMessage：

JAVA
/**
 * 发布确认模式
 * 1.单个确认
 * 2.批量确认
 * 3.异步批量确认
 */
public class ConfirmMessage {
    public static final int MESSAGE_COUNT = 1000;
    public static void main(String[] args) throws Exception {
        // 发布确认模式
        // 1.单个确认
        // 2.批量确认
        // 3.异步批量确认
        publishMessageAsync();
    }
    // 单个确认发布
    public static void publishMessageIndividually() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        // 队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, 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());
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功");
            }
        }
        // 结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布1000个消息单个确认耗时" + (end-begin) +"ms");
    }
    // 批量确认发布
    public static void publishMessageBatch() throws IOException, InterruptedException {
        Channel channel = RabbitMQUtils.getChannel();
        // 队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启确认发布
        channel.confirmSelect();
        // 开启时间
        long begin = System.currentTimeMillis();

        //批量确认消息的大小
        int batchSize = 1000;
        for (int i = 0; i< MESSAGE_COUNT; i++) {
            String message = i+"";
            channel.basicPublish("", queueName, null, message.getBytes());

            if (i%batchSize == 0) {
                // 发布确认
                channel.waitForConfirms();
            }
        }
        // 结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布1000个消息批量确认耗时" + (end-begin) +"ms");
    }
    // 异步确认发布
    public static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        // 队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启确认发布
        channel.confirmSelect();
        // 开启时间
        long begin = System.currentTimeMillis();

        /**
         * 线程安全有序的一个哈希表 适用于高并发的情况下
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目，只要给到序号
         * 3.支持高并发（多线程）
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

        // 消息确认成功的回调方法
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            // 如果是批量发布消息，则批量删除
            if (multiple) {
                // 2.删除掉已经确认的消息，剩下就是未确认的消息
                ConcurrentNavigableMap<Long, String> confirmedMap =
                        outstandingConfirms.headMap(deliveryTag);
                confirmedMap.clear();
            } else {
                // 单个确认
                outstandingConfirms.remove(deliveryTag);
            }

            System.out.println("确认的消息是：" + deliveryTag);
        };
        // 消息确认失败的回调方法
        /**
         * 1.消息的标识 deliveryTag
         * 2.是否批量 multiple
         */
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            String message = outstandingConfirms.get(deliveryTag);
            // 3.打印未确认的消息有哪些
            System.out.println("未确认的消息是：" + message + "未确认消息的标记是：" + deliveryTag);
        };

        // 准备消息的监听器 监听哪些消息成功了 哪些些消息失败了
        /**
         * 1.监听哪些消息成功了
         * 2.监听那些消息失败了
         */
        channel.addConfirmListener(ackCallback, nackCallback);  // 监听器有两种，一种只监听成功的，一种成功的和失败的都监听，这里的话是后者

        // 批量确认发布
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i+"";
            channel.basicPublish("", queueName, null, message.getBytes());
            // 1. 此处记录下所有要发送消息，消息的总和
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            // 发布确认 异步

        }
        // 结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布1000个消息异步确认发布耗时" + (end-begin) +"ms");
    }
}

5. 交换机

老模式：

新模式：

5.1 Exchange

5.1.1 Exchange的概念

RabbitMQ消息传递模型的核心思想：生产者生产的消息从不会直接发送到队列，甚至生产者也不知道消息传递到了哪个队列中。

生产者将消息发送给交换机，交换机会把接收的消息推入队列，交换机负责把这些消息放入指定的队列或者将其丢弃。其由交换机的类型决定。

5.1.2 无名Exchange

PLAINTEXT
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

第一个参数是交换机的名称，空字符串表示默认或无名交换机；消息能路由发送到队列其实是由routingKey（bindingkey）绑定key指定的。

5.2 临时队列

临时队列无名称。

创建临时队列的方式：

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

创建后的样子：

5.3 绑定队列

可以指定交换机绑定某个队列，通过绑定routingKey将消息传递给指定的队列

5.3.1 Fanout（扇出模式）

一个生产者发消息，多个消费者接收的都是一样的消息（随意广播），适用于发布订阅模式、打日志和多个系统共享等，消息会被转发到所有绑定到该交换机的队列上。

5.3.2 Fanout交换机实战

消费者1：

JAVA
public class Receiver1 {
    public static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        // 声明一个交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        // 声明一个临时队列，名称是随机的
        String queueName = channel.queueDeclare().getQueue();

        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println("R1等待接收消息，把接收到消息打印在控制台上");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("R1接收到的消息：" + new String(message.getBody()));
        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("R1取消消息");
        };
        channel.basicConsume(queueName, true, deliverCallback, cancelCallback);
    }
}

消费者2：

JAVA
public class Receiver2 {
    public static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        // 声明一个交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        // 声明一个临时队列，名称是随机的
        String queueName = channel.queueDeclare().getQueue();

        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println("R2等待接收消息，把接收到消息打印在控制台上");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("R2接收到的消息：" + new String(message.getBody()));
        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("R2取消消息");
        };
        channel.basicConsume(queueName, true, deliverCallback, cancelCallback);
    }
}

生产者：

JAVA
public class Emit {
    public static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String message = scanner.next();
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
            System.out.println("生产者发出消息：" + message);
        }
    }
}

5.3.3 Direct exchange

一个生产者发消息可以指定消费者接收某个消息或者不接收某个消息，当然也可以指定所有消费者都接收一样的消息（有选择性接收日志）

直接绑定但绑定的多个队列key相同

5.3.4 Direct交换机实战

消费者1：

JAVA
public class Receiver1 {
    public static final String EXCHANGE_NAME = "direct";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        channel.queueDeclare("console", false, false, false, null);
        channel.queueBind("console", EXCHANGE_NAME, "info");
        channel.queueBind("console", EXCHANGE_NAME, "warning");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("R1接收到的消息：" + new String(message.getBody()));
        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("R1取消消息");
        };
        channel.basicConsume("console", true, deliverCallback, cancelCallback);
    }
}

消费者2：

JAVA
public class Receiver2 {
    public static final String EXCHANGE_NAME = "direct";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        channel.queueDeclare("disk", false, false, false, null);
        channel.queueBind("disk", EXCHANGE_NAME, "error");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("R2接收到的消息：" + new String(message.getBody()));
        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("R2取消消息");
        };
        channel.basicConsume("console", true, deliverCallback, cancelCallback);
    }
}

生产者：

PLAINTEXT
public class Emit {
    public static final String EXCHANGE_NAME = "direct";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String message = scanner.next();
            channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes());
            System.out.println("生产者发送消息：" + message);
        }
    }
}

5.3.5 多重绑定

PLAINTEXT
channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes());

routingKey决定把消息发送给哪个消费者，上面是把消息发送给绑定了info的队列

5.4 Topic交换机

5.4.1 前几种交换机模式的缺陷：

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

5.4.2 Topic交换机对routingKey的书写要求

发送到类型是topic交换机的消息的routingkey不能随意写，必须满足一定的要求，它必须是一个单
词列表，以点号分隔开。这些单词可以是任意单词，比如说：”stock.usd.nyse”,”nye.mw”,
‘quick.orange.rabbit’”.这种类型的。当然这个单词列表最多不能超过255个字节。
在这个规则列表中，其中有两个替换符是大家需要注意的
*(星号)可以代替一个单词
#(井号)可以替代零个或多个单词

5.4.3 Topic匹配案例

如果有一个消息符合同一个队列的两个绑定条件，但是它只会被该队列接收一次。如果有消息不符合任意一个绑定条件，则会被丢弃。

5.4.4 Topic交换机实战：

消费者1：

JAVA
public class ReceiverTopic1 {
    public static final String EXCHANGE_NAME = "topic";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = "Q1";
        channel.queueDeclare(queueName, false, false, false, null); // 不持久化，不共享，不自动删除，无参数
        channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*");
        System.out.println("Q1等待接收消息");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Q1接收到的消息：" + new String(message.getBody()) +
                    "接收队列：" + queueName + "绑定键：" + message.getEnvelope().getRoutingKey());
            System.out.println("好好好");
        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("Q1取消消息");
        };
        channel.basicConsume(queueName, deliverCallback, cancelCallback);

    }
}

消费者2：

JAVA
public class ReceiverTopic2 {
    public static final String EXCHANGE_NAME = "topic";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        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("Q2等待接收消息");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Q2接收到的消息：" + new String(message.getBody()) +
                    "接收队列：" + queueName + "绑定键：" + message.getEnvelope().getRoutingKey());
        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("Q2取消消息");
        };
        channel.basicConsume(queueName, deliverCallback, cancelCallback);

    }
}

生产者：

PLAINTEXT
public class EmitTopic {
    public static final String EXCHANGE_NAME = "topic";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();

        Map<String, String> bindingKeyMap = new HashMap<>();
        bindingKeyMap.put("quick.orange.rabbit","被队列Q1Q2接收到");
        bindingKeyMap.put("lazy.orange.elephant'","被队列Q1Q2接收到");
        bindingKeyMap.put("quick.orange.fox","被队列Ql接收到");
        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 routineKey = bindingKeyEntry.getKey();
            String message = bindingKeyEntry.getKey();
            channel.basicPublish(EXCHANGE_NAME, routineKey,null, message.getBytes("UTF-8"));
            System.out.println("生产者发出消息：" + message);
        }

    }
}

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

6. 死信队列

6.1 死信的概念

何为死信？顾名思义就是无法被消费的消息。一般来说，生产者 将消息投递到broker或者直接到queue里，消费者从中取出消息进行消费，但有时由于某些原因导致消息不能被消费，这样的消息如果没有后续的处理就会变成死信，那么存放死信的队列就是死信队列。

应用场景：

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

6.2 产生死信的原因

1. 消息TTL过期
2. 队列达到最大长度
3. 消息被拒绝并且requeue = false（不予重新入队）

6.3 死信实战

6.3.1 代码架构图

死信队列代码演示：

Consumer01:

JAVA
public class Consumer01 {
    // 普通交换机名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    // 死信交换机名称
    public static final String DEAD_EXCHANGE = "dead_exchange";
    // 普通队列名称
    public static final String NORMAL_QUEUE = "normal_queue";
    // 死信队列名称
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, "direct");
        channel.exchangeDeclare(DEAD_EXCHANGE, "direct");
        // 声明普通队列
        Map<String, Object> arguments = new HashMap<>();
        // 过期时间
        // arguments.put("x-message-ttl", 10000);
        // 正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信routingKey
        arguments.put("x-dead-letter-routing-key", "lisi");

        channel.queueDeclare(NORMAL_QUEUE, false, false, false, null);
        // 声明死信队列
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);

        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");
        System.out.println("等待接收消息");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("consumer01接收的消息：" + new String(message.getBody()));

        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("Q1取消消息");
        };
        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, cancelCallback);
    }
}

Producer:

JAVA
/**
 * 死信队列之生产者
 */
public class Producer {
    // 普通交换机名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        //死信消息 设置TTL时间
        AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder().expiration("10000").build();

        // 延迟消息
        for (int i = 0;i < 10;i++) {
            String message = i + "";
            channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", null, message.getBytes());
        }
    }
}

Consumer02:

JAVA
public class Consumer02 {
    // 死信队列名称
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        System.out.println("consumer02等待接收消息");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("consumer02接收的消息：" + new String(message.getBody()));

        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("C2取消消息");
        };
        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, cancelCallback);
    }
}

6.3.2 消息TTL过期

JAVA
// 方式1
//死信消息 设置TTL时间
AMQP.BasicProperties properties = new AMQP.BasicProperties()
        .builder().expiration("10000").build();
// 方式2
Map<String, Object> arguments = new HashMap<>();
// 过期时间
arguments.put("x-message-ttl", 10000);

6.3.3 队列达到最大长度

PLAINTEXT
public class Consumer01 {
    // 普通交换机名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    // 死信交换机名称
    public static final String DEAD_EXCHANGE = "dead_exchange";
    // 普通队列名称
    public static final String NORMAL_QUEUE = "normal_queue";
    // 死信队列名称
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        // 声明死信和普通交换机，类型为direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 声明普通队列
        Map<String, Object> arguments = new HashMap<>();
        // 过期时间
        // arguments.put("x-message-ttl", 10000);
        // 正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信routingKey
        arguments.put("x-dead-letter-routing-key", "lisi");
        // 设置正常队列的长度限制
        arguments.put("x-max-length", 6);

        // 声明普通队列
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, null);
        // 声明死信队列
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);

        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");
        System.out.println("consumer01等待接收消息");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("consumer01接收的消息：" + new String(message.getBody()));

        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("C1取消消息");
        };
        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, cancelCallback);
    }
}

6.3.4 消息被拒

PLAINTEXT
DeliverCallback deliverCallback = (consumerTag, message) -> {
    String msg = new String(message.getBody(), "UTF-8");
    if (msg.equals("info5")) {
        System.out.println(msg + "：此消息是被拒绝的");
        channel.basicReject(message.getEnvelope().getDeliveryTag(), false); //拒绝此消息并不放回普通队列
    }
    System.out.println("consumer01接收的消息：" + new String(message.getBody()));
};

7. 延迟队列（死信队列实现）

7.1 延迟队列的概念

延迟队列的内部是有序的，最重要的属性体现在它的延迟属性上。延迟队列就是用来存放需要在指定时间被处理的元素的队列。

7.2 延迟队列的使用场景

1. 订单在十分钟内未支付则取消
2. 新创建的店铺，如果在十天内未上传过商品，则自动发送消息提醒
3. 用户注册成功后，如果三天内没有登录则发消息提醒
4. 用户发起退款，如果三天内没有得到处理则通知相关运用那个人员
5. 预定会议后，在预定时间前十分钟通知参会人员参加会议

7.3 延迟队列的代码实现（整合Spring Boot）

TtlQueueConfig:

PLAINTEXT
@Configuration
public class TtlQueueConfig {
    // 普通交换机名称
    public static final String X_EXCHANGE = "X";
    // 死信交换机名
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    // 普通队列名称
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    // 死信队列名称
    public static final String DEAD_LETTER_QUEUE = "QD";

    // 声明xExchange
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }

    // 声明yExchange
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    //声明普通队列  要有ttl 为10s
    @Bean("queueA")
    public Queue queueA() {
        Map<String, Object> arguments = new HashMap<>();
        // 设置死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        arguments.put("x-dead-letter-routing-key", "YD");
        arguments.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
    }

    //声明普通队列  要有ttl 为40s
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> arguments = new HashMap<>(3);
        // 设置死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        arguments.put("x-dead-letter-routing-key", "YD");
        arguments.put("x-message-ttl", 40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
    }

    //声明死信队列  要有ttl 为40s
    @Bean("queueD")
    public Queue queueD() {
        return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
    }

    //声明队列 QA 绑定 X 交换机
    @Bean
    public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }
    //声明队列 QB 绑定 X 交换机
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    }
    //声明队列 QD 绑定 Y 交换机
    @Bean
    public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,
                                  @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }
}

DeadLetterQueueConsumer:

PLAINTEXT
	@Slf4j
@Component
public class DeadLetterQueueConsumer {
    // 接收消息
    @RabbitListener(queues = "QD")
    public void receiveD(Message message, Channel channel)throws Exception {
        String msg = new String(message.getBody());
        log.info("当前时间：{}，收到死信队列的消息：{}", new Date().toString(), msg);
    }
}

SendMessageController:

PLAINTEXT
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @GetMapping("/sendMsg/{message}")
    public void sendMsg(@PathVariable String message) {
        rabbitTemplate.convertAndSend("X", "XA", "消息来自ttl为10s的队列：" + message);
        rabbitTemplate.convertAndSend("X", "XB", "消息来自ttl为40s的队列：" + message);
        log.info("当前时间：{}，发送一条消息给两个TTL队列：{}", new Date().toString(), message);
    }
}

7.4 延迟队列的优化

新增一个队列QC，绑定关系如上，但不设置TTL。

TtlQueueConfig:

PLAINTEXT
@Configuration
public class TtlQueueConfig {
    // 普通交换机名称
    public static final String X_EXCHANGE = "X";
    // 死信交换机名
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    // 普通队列名称
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    // 死信队列名称
    public static final String DEAD_LETTER_QUEUE = "QD";
    // 普通队列的名称
    public static final String QUEUE_C = "QC";

    // 声明QC队列
    @Bean("queueC")
    public Queue queueC(){
        Map<String ,Object> arguments = new HashMap<>(3);
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE); // 设置死信交换机
        arguments.put("x-dead-routing-key", "YD"); // 设置死信routingKey
        // 设置TTL
        return QueueBuilder.durable().withArguments(arguments).build();
    }

    @Bean
    public Binding queueCBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueC).to(xExchange).with("XC");
    }

    // 声明xExchange
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }

    // 声明yExchange
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    //声明普通队列  要有ttl 为10s
    @Bean("queueA")
    public Queue queueA() {
        Map<String, Object> arguments = new HashMap<>();
        // 设置死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        arguments.put("x-dead-letter-routing-key", "YD");
        arguments.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
    }

    //声明普通队列  要有ttl 为40s
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> arguments = new HashMap<>(3);
        // 设置死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        arguments.put("x-dead-letter-routing-key", "YD");
        arguments.put("x-message-ttl", 40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
    }

    //声明死信队列  要有ttl 为40s
    @Bean("queueD")
    public Queue queueD() {
        return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
    }

    //声明队列 QA 绑定 X 交换机
    @Bean
    public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }
    //声明队列 QB 绑定 X 交换机
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    }
    //声明队列 QD 绑定 Y 交换机
    @Bean
    public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,
                                  @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }
}

SendMessageController:

PLAINTEXT
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @GetMapping("/sendMsg/{message}")
    public void sendMsg(@PathVariable String message) {
        rabbitTemplate.convertAndSend("X", "XA", "消息来自ttl为10s的队列：" + message);
        rabbitTemplate.convertAndSend("X", "XB", "消息来自ttl为40s的队列：" + message);
        log.info("当前时间：{}，发送一条消息给两个TTL队列：{}", new Date().toString(), message);
    }

    @GetMapping("/sendExpirationMsg/{message}/{ttlTime}")
    public void sendMsg(@PathVariable String message, @PathVariable String ttlTime){
        rabbitTemplate.convertAndSend("X", "XC", message, msg -> {
            msg.getMessageProperties().setExpiration(ttlTime);
            return msg;
        });
        log.info("当前时间：{}，发送一条时长{}毫秒TTL消息给两个TTL队列QC：{}", new Date().toString(), ttlTime, message);
    }
}

缺陷：

如果发送的第一个消息msg1的TTL为20s，第二个消息msg2的TTL为2s，死信队列先收到的是msg1而不是msg2。因为RabbitMQ只会检查第一个消息是否过期如果过期则丢到死信队列，如果第一个消息的诞时时长很而第二个消息的延时时长很短第二个消息并不会优先得到执行。

7.5 延迟队列插件实现

基于插件的延迟队列的配置类：

JAVA
@Configuration
public class DelayedQueueConfig {
    // 交换机
    public static final String DELAYED_QUEUE_NAME = "delayed.queue";
    // 队列
    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
    // routingKey
    public static final String DELAYED_ROUTINGKEY = "delayed.routingKey";

    @Bean
    public Queue delayedQueue() {
        return new Queue(DELAYED_QUEUE_NAME);
    }

    // 声明交换机
    @Bean
    public CustomExchange delayedExchange() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-delayed-type", "direct");
        /**
         * 1.交换机的名称
         * 2.交换机的类型
         * 3.是否需要自动化
         * 4.是否需要自动删除
         */
        return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false);
    }

    // 绑定交换机和队列
    @Bean
    public Binding delayedQueueBindingDelayedExchange(
            @Qualifier("delayedQueue") Queue delayedQueue,
            @Qualifier("delayedExchange") CustomExchange delayedExchange) {
        return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTINGKEY).noargs();
    }
}

生产者：

JAVA
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @GetMapping("/sendMsg/{message}")
    public void sendMsg(@PathVariable String message) {
        rabbitTemplate.convertAndSend("X", "XA", "消息来自ttl为10s的队列：" + message);
        rabbitTemplate.convertAndSend("X", "XB", "消息来自ttl为40s的队列：" + message);
        log.info("当前时间：{}，发送一条消息给两个TTL队列：{}", new Date().toString(), message);
    }
	// 发送基于死信队列的延迟消息
    @GetMapping("/sendExpirationMsg/{message}/{ttlTime}")
    public void sendMsg(@PathVariable String message, @PathVariable String ttlTime){
        rabbitTemplate.convertAndSend("X", "XC", message, msg -> {
            msg.getMessageProperties().setExpiration(ttlTime);
            return msg;
        });
        log.info("当前时间：{}，发送一条时长{}毫秒TTL消息给两个TTL队列QC：{}", new Date().toString(), ttlTime, message);
    }
	// 发送基于插件的延迟队列的消息
    @GetMapping("/sendDelayMsg/{message}/{delayTime}")
    public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime) {
        rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME,
                DelayedQueueConfig.DELAYED_ROUTINGKEY, message, msg -> {
            // 发送消息的时候 延迟时长 单位ms
            msg.getMessageProperties().setDelay(delayTime);
            return msg;
        });
        log.info("当前时间：{}，发送一条时长{}毫秒的消息给延迟队列delayed.queue:{}", new Date().toString(), delayTime, message);
    }
}

消费者：

JAVA
@Component
@Slf4j
public class DelayedQueueConsumer {

    @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
    public void receiveDelayQueue(Message message) {
        String msg = new String(message.getBody());
        log.info("当前时间：{}，收到延迟队列的消息：{}", new Date().toString(), msg);
    }
}

总结：

延时队列在需要延时处理的场景下非常有用，使用RabbitMQ来实现延时队列可以很好的利用
RabbitMQ的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过RabbitMQ集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。
当然，延时队列还有很多其它选择，比如利用Java的DelayQueue,利用Redis的set,利用Quartz
或者利用kafka的时间轮，这些方式各有特点看需要适用的场景

8. 发布确认高级

RabbitMQ会由于某些原因导致重启，那么生产者投递的消息无法被消费导致消息丢失，需要手动处理和恢复。在RabbitMQ集群不可用的时候，怎么处理无法投递的消息呢？

8.1 发布确认Spring Boot版本

8.1.1 确认机制方案

生产者发送消息给交换机，但交换机无法接受消息，则把消息放在缓存中。当交换机能收到消息时通过定时任务把缓存中的消息重新投递到交换机，并清除缓存中的消息。

8.1.2 代码结构图

8.1.3 编写配置类（注意一定要加上@Configuration注解）

JAVA
/**
 * 发布确认高级配置类
 */
@Configuration
public class ConfirmSeniorConfig {
    // 交换机
    public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
    // 队列
    public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
    // routingKey
    public static final String CONFIRM_ROUTING_KEY = "key1";

    // 声明交换机
    @Bean("confirmExchange")
    public DirectExchange confirmExchange() {
        return new DirectExchange(CONFIRM_EXCHANGE_NAME);
    }

    // 声明队列
    @Bean("confirmQueue")
    public Queue confirmQueue() {
        return new Queue(CONFIRM_QUEUE_NAME);
    }

    // 绑定交换机和队列
    @Bean
    public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
                                        @Qualifier("confirmExchange") DirectExchange confirmExchange) {
        return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
    }
}

生产者：

PLAINTEXT
/**
 * 发布确认高级生产者
 */
@RestController
@Slf4j
@RequestMapping("/confirmSenior")
public class ProducerController {
    @Autowired
    RabbitTemplate rabbitTemplate;
    // 发消息
    @GetMapping("/sendMessage/{message}")
    public void sendMessage(@PathVariable String message) {
        CorrelationData correlationData = new CorrelationData("1");
        rabbitTemplate.convertAndSend(ConfirmSeniorConfig.CONFIRM_EXCHANGE_NAME,
                ConfirmSeniorConfig.CONFIRM_ROUTING_KEY, message, correlationData);
        log.info("发送消息内容为{}", message);
    }
}

消费者：

PLAINTEXT
/**
 * 发布确认高级消费者
 */
@Slf4j
@Component
public class ConfirmSeniorConsumer {
     @RabbitListener(queues = ConfirmSeniorConfig.CONFIRM_QUEUE_NAME)
    public void receiveConfirmMessage(Message message) {
         String msg = new String(message.getBody());
         log.info("接受到的队列confirm.queue的消息：{}", msg);
     }
}

8.1.4 回调接口

当生产者发送消息不能被交换机接收时，会出发回调接口的方法。谁发送消息谁实现回调接口。

调用回调方法的时机：

1. 发消息给交换机，交换机接收到了 => 触发回调

   重写方法各个形参的代表意思

   1. correlationData保存回调消息的id和相关信息
   2. 交换机收到消息 true
   3. cause 原因

2. 发消息给交换机，交换机接收失败 => 触发回调

   1. correlationData 保存回调消息的id和相关信息
   2. 交换机收到消息 => false
   3. cause 接收消息失败的原因

PLAINTEXT
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {
    @Autowired
    RabbitTemplate rabbitTemplate;
    // 将重写的方法注入到RabbitTemplate.ConfirmCallback这个接口中
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
    }
    /**
     * 交换机确认回到方法
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("交换机已经收到Id为：{}的消息", id);
        } else {
            log.info("交换机还未收到Id为：{}消息，由于原因：{}", id, cause );
        }
    }
}

注意：

1. 实现RabbitTemplate.ConfirmCallback接口时一定要将重写的confirm方法重新注入到RabbitTemplate.ConfirmCallback接口中，不然重写的方法不会被自动调用

   1. 注意注入的顺序 实现RabbitTemplate.ConfirmCallback接口的类先注入，再是RabbitTemplate，最后才是重写的Confirm方法

2. 调用回调接口要在配置文件中，添加

   JAVA
   spring.rabbitmq.publisher-confirm-type=correlated

   

   SIMPLE模式具有关闭信道channel的风险

3. 如果不重写回调接口，交换机就不知道消息接收成功与否，这样子也难以将发送失败的消息给临时存储起来

4. 生产者发送消息时可以指定发送消息的id

   PLAINTEXT
   CorrelationData correlationData = new CorrelationData("1");

交换机接收成功：

交换机接收失败：

如果生产者发送的消息是这样的：（即发送一个正确routingKey的消息和发送一个错误routingKey的消息）。这样子的话队列接收到第一个信息，但未接收到第二个消息，并且队列不会应答，只有交换机确认了。

PLAINTEXT
// 发消息
@GetMapping("/sendMessage/{message}")
public void sendMessage(@PathVariable String message) {
    CorrelationData correlationData1 = new CorrelationData("1");
    rabbitTemplate.convertAndSend(ConfirmSeniorConfig.CONFIRM_EXCHANGE_NAME,
            ConfirmSeniorConfig.CONFIRM_ROUTING_KEY, message, correlationData1);
    log.info("发送消息内容为{}", message);

    CorrelationData correlationData2 = new CorrelationData("2");
    rabbitTemplate.convertAndSend(ConfirmSeniorConfig.CONFIRM_EXCHANGE_NAME,
            ConfirmSeniorConfig.CONFIRM_ROUTING_KEY + "2", message, correlationData2);
    log.info("发送消息内容为{}", message);

}

结果如下：

8.2 回退消息

8.1.1 Mandatory参数

在仅开启生产者确认机制的情况下，交换机接收到消息后会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息则会被直接丢弃，而生产者是不知道消息被丢弃了。

如何回退消息？ => 实现RabbitTemplate.ReturnsCallback接口的returnedMessage回退方法

• 在配置文件中添加如下配置，开启消息回退。

  PLAINTEXT
  spring.rabbitmq.publisher-returns=true

• 实现RabbitTemplate.ReturnsCallback接口的回调方法

  PLAINTEXT
  @Component
  @Slf4j
  public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
      @Autowired
      RabbitTemplate rabbitTemplate;
      // 将重写的方法注入到RabbitTemplate.ConfirmCallback这个接口中
      @PostConstruct
      public void init() {
          rabbitTemplate.setConfirmCallback(this);
          
      }
      /**
       * 交换机确认消息的回调方法（不管消息是否接收成功都会调用这个方法）
       * @param correlationData
       * @param ack
       * @param cause
       */
      @Override
      public void confirm(CorrelationData correlationData, boolean ack, String cause) {
          String id = correlationData != null ? correlationData.getId() : "";
          if (ack) {
              log.info("交换机已经收到Id为：{}的消息", id);
          } else {
              log.info("交换机还未收到Id为：{}消息，由于原因：{}", id, cause );
          }
      }
  
      // 回退方法 （在消息在传递的过程中不可到达目的地时将消息返回给生产者）
      @Override
      public void returnedMessage(ReturnedMessage returnedMessage) {
          log.info("消息{}，被交换机{}退回，回退原因：{}，路由key：{}", returnedMessage.getMessage(),
                  returnedMessage.getExchange(), returnedMessage.getReplyText(), returnedMessage.getRoutingKey());
      }
  }

  重现发送消息发现并没有调用回退方法，因为未把重写的回退方法重新注入至那个实现的接口

  

重新注入就好了：

PLAINTEXT
@PostConstruct
public void init() {
    rabbitTemplate.setConfirmCallback(this);
    rabbitTemplate.setReturnsCallback(this);
}

成功示例：

完整代码：

PLAINTEXT
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
    @Autowired
    RabbitTemplate rabbitTemplate;
    // 将重写的方法注入到RabbitTemplate.ConfirmCallback这个接口中
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }
    /**
     * 交换机确认消息的回调方法（不管消息是否接收成功都会调用这个方法）
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("交换机已经收到Id为：{}的消息", id);
        } else {
            log.info("交换机还未收到Id为：{}消息，由于原因：{}", id, cause );
        }
    }

    // 回退方法 （在消息在传递的过程中不可到达目的地时将消息返回给生产者）
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        log.info("消息{}，被交换机{}退回，回退原因：{}，路由key：{}", returnedMessage.getMessage(),
                returnedMessage.getExchange(), returnedMessage.getReplyText(), returnedMessage.getRoutingKey());
    }
}

8.3 备份交换机

8.3.1 代码架构图

8.3.2 修改配置类

在配置类中多加一个备份交换机、备份队列、报警队列、备份消费者（可不写）和报警消费者。注意还要说明确认交换机无法接收消息时，需将消息转发给备份交换机。并把之前创建的确认交换机给删除，因为它的信息被修改了。

结果分析：

8.3.3 总结

Mandatory参数和备份交换机可以一起使用，但同时开启Mandatory和备份交换机，消息会到哪里？

实验结果：

老师的实验结果是备份交换机优先级更高，但是我的结果却是消息被回退了和老师不一致。（根据自己的实际场景来）

9. RabbitMQ其他知识点

9.1 幂等性

9.1.1 概念

什么是幂等性？幂等性指的是无论执行多少次同样的操作，结果都是一致的。举个最简单的例子，那就是支付，用户购买商品后支付，支付扣款成功，但是返回结果的时候网络异常，此时钱已经扣了，用户再次点击按钮，此时会进行第二次扣款，返回结果成功，用户查询余额发现多扣钱了，流水记录也变成了两条。在以前的单应用系统中，我们只需要把数据操作放入事务中即可，发生错误立即回滚，但是再响应客户端的时候也有可能出现网络中断或者异常等等

9.1.2 消息重复消费

消费者在消费MQ中的消息时，MQ已经把消息发给消费者，但消费者在给MQ返回ack时网络中断，故MQ未收到确认消息，导致该消息会在网络正常时再次发给消费者，导致消费者消费了重复的消息。

9.1.3 解决思路1（乐观锁）

MQ消费者的幂等性的解决一般使用全局ID或者写个唯一标识比如时间戳或者UUID或者订单消费者消费MQ中的消息也可利用MQ的该id来判断，或者可按自己的规侧则生成一个全局唯一id,每次消费消息时用该d先判断该消息是否已消费过。

9.1.4 解决思路2 消费端的幂等性保障

在海量订单生成的业务高峰期，生产端有可能就会重复发生了消息，这时候消费端就要实现幂等性，这就意味着我们的消息永远不会被消费多次，即使我们收到了一样的消息。业界主流的幂等性有两种操作：a.唯一ID+指纹码机制，利用数据库主键去重，b.利用redis的原子性去实现

9.1.5 解决思路3 唯一ID+指纹码机制

指纹码：我们的一些规侧或者时间戳加别的服务给到的唯一信息码，它并不一定是我们系统生成的，基本都是由我们的业务规则拼接而来，但是一定要保证唯一性，然后就利用查询语句进行判断这个d是否存在数据库中，优势就是实现简单就一个拼接，然后查询判断是否重复；劣势就是在高并发时，如果是单个数据库就会有写入性能瓶颈当然也可以采用分库分表提升性能，但也不是我们最推荐的方式。

9.1.6 解决思路4 Redis原子性（推荐）

利用redis执行setnx命令，天然具有幂等性。从而实现不重复消费

9.2 优先级队列

优先级队列的优先级范围为0~255，越大越先执行

9.2.1 使用场景

在我们系统中有一个订单催付的场景，我们的客户在天猫下的订单，淘宝会及时将订单推送给我们，如
果在用户设定的时间内未付款那么就会给用户推送一条短信提醒，很简单的一个功能对吧，但是，tm商家对我们来说，肯定是要分大客户和小客户的对吧，比如像苹果，小米这样大商家一年起码能给我们创造很大的利润，所以理应当然，他们的订单必须得到优先处理，而曾经我们的后端系统是使用ds来存放的定时轮询，大家都陈知道redis只能用List做一个简简单单的消息队列，并不能实现一个优先级的场景，所以订单量大了后采用RabbitMQ进行改造和优化如果发现是大客户的订单给一个相对比较高的优先级，否则就是默认优先级。

9.2.2 如何添加优先级队列

1. 控制台添加

   

2. 队列中代码添加优先级

   

3. 消息中代码添加优先级

   

生产者：

PLAINTEXT
/**
 * 生产者发消息
 * @author 17653
 */
public class Producer1 {
    // 队列名称
    public static final String QUEUE_NAME = "hello";

    // 发消息
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置工厂 IP 连接RabbitMQ的队列
        factory.setHost("localhost");
        // 用户名和密码
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 获取信道
        // 创建连接
        Connection connection = factory.newConnection();
        // 获取信道
        Channel channel = connection.createChannel();
        /**
         * 生成一个队列
         * 1.队列名称
         * 2.队列中消息是否持久化（存在磁盘中） 默认情况下消息存在内容中
         * 3.该队列是否只供一个消费者消费 （是否进行消息共享）true：可以多个消费者消费  false：只能一个消费者消费
         * 4.是否自动删除 最后一个消费者端开连接以后该队一句是否自动删除true自动删除false不自动删除
         * 5.其它参数
         */
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-max-priority", 10); // 官方允许是0~255之间 此处优先级上限为10，防止浪费CPU和内存
        channel.queueDeclare(QUEUE_NAME, false, false, false, arguments);
        // 发消息
        for (int i = 1;i < 11;i++) {
            String message = "info" + i;
            if (i == 5) {
                AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();
                channel.basicPublish("", QUEUE_NAME, properties, message.getBytes());
            } else {
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            }
        }
        /**
         * 发送一个消费
         * 1.发送到哪个交换机
         * 2.路由的Ky值是哪个 本次是队列的名称
         * 3.其它参数信息
         * 4.发送消息的本体
         */
        System.out.println("消息发送完毕");
    }
}

消费者：（消费者并没有任何改变，以前该怎么接收就怎么接收）

PLAINTEXT
public class Consumer1 {
    // 队列的名称
    public static final String QUEUE_NAME = "hello";
    // 接收消息
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        // 创建信道
        Channel channel = connection.createChannel();

        // 声明 接收消息
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println(message);
        };

        // 声明 取消消息时的回调
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消息消费被中断");
        };
        /**
         * 消费者消费信息
         * 1.消费哪个队列
         * 2.消费成功后是否要应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费的回调
         */

         channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}

结果如下：

注意事项：

要让队列实现优先级需要做的事情有如下事情：队列需要设置为优先级队列，消息需要设置消息的优先
级，消费者需要等待消息已经发送到队列中才去消费，因为这样才有机会对消息进行排序

9.3 惰性队列

惰性队列：消息保存在内存中还是磁盘中

正常情况：消息保存在内存中

惰性队列：消息保存在磁盘中，消费速度慢

使用场景

RabbitMQ从3.6.0版本开始引入了惰性队列的概念。惰性队列会尽可能的将消息存入磁盘中，而在消费者消费到相应的消息时才会被加载到内存中，它的一个重要的设计目标是能够支持更长的队列，即支持更多的消息存储。当消费者由于各种各样的原因（比如消费者下线、宕机亦或者是由于维护而关闭等）而致使长时间内不能消费消息造成堆积时，惰性队列就很有必要了。默认情况下，当生产者将消息发送到RabbitMQ的时候，队列中的消息会尽可能的存储在内存之中，这样可以更加快速的将消息发送给消费者。即使是持久化的消息，在被写入磁盘的同时也会在内存中驻留一份备份。当RabbitMQ需要释放内存的时候，会将内存中的消息换页至磁盘中，这个操作会耗费较长的时间，也会阻塞队列的操作，进而无法接收新的消息。虽然RabbitMQ的开发者们一直在升级相关的算法，但是效果始终不太理想，尤其是在消息量特别大的时候。

（消费者宕机导致大量消息无非消费全部堆积在队列中，这时可以用惰性队列）

9.3.1 两种模式

队列具有两种模式：default和lazy。

模式即为惰性队列的模式，可以通过调用channel.queue Declare方法的时候在参数中设置，也可以通过
Policy的方式设置，如果一个队列同时使用这两种方式设置的话，那么Policy的方式具备更高的优先级。
如果要通过声明的方式改变已有队列的模式的话，那么只能先删除队列，然后再重新声明一个新的。

在队列声明的时候可以通过x-queue-mode”参数来设置队列的模式，取值为”default’和azy’。下面示
例中演示了一个惰性队列的声明细节：

PLAINTEXT
Map<String,Object>args new HashMap<String,Object>()
args.put("x-queue-mode","lazy");
channel.queueDeclare("myqueue",false,false,false,args);

或者在控制台设置：

9.3.2 内存开销对比

在发送一百万条消息，每条消息大概占1KB的情况下，普通队列占用内存1.2GB，而惰性队列仅仅占用1.5MB

10. RabbitMQ集群

10.1 搭建Rabbit MQ集群

10.2 镜像队列

在创建RabbitMQ集群时，如果1号MQ宕机，那么一号MQ的队列就会被备份到其他任一一台MQ上，比如备份到2号MQ上，计算2号MQ宕机了其还会备份至3号机上。

控制台Policy设置镜像队列：

10.3 Haproxy + keepalive实现高可用负载均衡

10.3.1 整体架构图

10.3.2 Haproxy实现负载均衡

HAProxy提供高可用性、负载均衡及基于TCPHTTP应用的代理，支持虚拟主机，它是免费、快速并
且可靠的一种解决方案，包括Twitter,Reddit,StackOverflow,GitHub在内的多家知名互联网公司在使用。
HAProxy实现了一种事件驱动、单一进程模型，此模型支持非常大的井发连接数。

10.4 Federation Exchange（联邦交换机）

10.4.1 使用原因

(broker北京)，(broker深圳)彼此之间相距甚远，网络延迟是一个不得不面对的问题。有一个在北京的业务(Client北京)需要连接(broker北京)，向其中的交换器exchangeA发送消息，此时的网络延迟很小，(Client:北京)可以迅速将消息发送至exchangeA中，就算在开启了publisherconfirm机制或者事务机制的情况下，也可以迅速收到确认信息。此时又有个在深圳的业务(Client深圳)需要向exchangeA发送消息，那么(Client深圳)(broker北京)之间有很大的网络延迟，(Client深圳)将发送消息至exchangeA会经历一定的延迟，尤其是在开启了publisherconfirm机制或者事务机制的情况下，(Client深圳)会等待很长的延迟时间来接(broker北京)的确认信息，进而必然造成这条发送线程的性能降低，甚至造成一定程度上的阻塞。
将业务(Client深圳)部署到北京的机房可以解决这个问题，但是如果(Client深圳)调用的另些服务都部
署在深圳，那么又会引发新的时延问题，总不见得将所有业务全部部署在一个机房，那么又该如何实现？
这里使用Federation插件就可以很好地解决这个问题，

10.4.2 搭建步骤

1.需要保证每台节点单独运行
2.在每台机器上开启federation相关插件

PLAINTEXT
rabbitmq-plugins enable rabbitmq_federation
rabbitmq-plugins enable rabbitmq_federation_management

3.原理图（先运行consumer在node2创建fed_exchange)

10.5 Shovel（数据迁移）

10.5.1 使用原因

Federation具备的数据转发功能类似，Shovel够可靠、持续地从一个Broker中的队列（作为源端，即
source)拉取数据并转发至另一个Broker中的交换器（作为目的端，即destination)。作为源端的队列和作
为目的端的交换器可以同时位于同一个Broker,也可以位于不同的Broker上。Shovel可以翻译为”铲子”，
是一种比较形象的比喻，这个”铲子”可以将消息从一方”铲子”另一方。Shovel行为就像优秀的客户端应用
程序能够负责连接源和目的地、负责消息的读写及负责连接失败问题的处理。

10.6.2 搭建步骤

1. 开启插件

PLAINTEXT
rabbitmq-plugins enable rabbitmq_shovel
rabbitmq-plugins enable rabbitmq_shovel_management

1. 原理图（在源头发送的消息直接进入到目的队列）