RabbitMQ

什么是MQ

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

RabbitMQ是一个消息中间件:它接受并转发消息。你可以把它当做一个快递站点，当你要发送一个包裹时，你把你的包裹放到快递站，快递员最终会把你的快递送到收件人那里，按照这种逻辑RabbitMQ是一个快递站，一个快递员帮你传递快件。RabbitMQ与快递站的主要区别在于，它不处理快件而是接收，存储和转发消息数据。

为什么要用MQ

1. 流量消峰

   举个例子，如果订单系统最多能处理一万次订单，这个处理能力应付正常时段的下单时绰绰有余，正常时段我们下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的，只能限制订单超过一万后不允许用户下单。使用消息队列做缓冲，我们可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这时有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验要好

2. 应用解耦

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

3. 异步处理

   有些服务间调用是异步的，例如A调用B,B需要花费很长时间执行，但是A需要知道B什么时候可以执行完，以前一般有两种方式，A过一段时间去调用B的查询api查询。或者A提供一个callback api,B执行完之后调用api通知A服务。这两种方式都不是很优雅，使用消息总线，可以很方便解决这个问题,A调用B服务后，只需要监听B处理完成的消息，当B处理完成后，会发送一条消息给MQ，MQ会将此消息转发给A服务。这样A服务既不用循环调用B的查询api，也不用提供callback api。同样B服务也不用做这些操作。A服务还能及时的得到异步处理成功的消息。

四大核心概念

• 生产者

  产生数据发送消息的程序是生产者

• 交换机

  交换机是RabbitMQ非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定

• 队列

  队列是RabbitMQ内部使用的一种数据结构，尽管消息流经RabbitMQ和应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式

• 消费者

架构

• Producer：消息生产者，即消息的发布方
• Connect：客户端与Broker间的TCP连接
• Channel：信道，每个连接采用多路复用，包含多个信道。producer与broker间采用信道传递数据
• Broker：Rabbit服务节点
• Vhost：虚机机，一个节点下包含多个vhost，vhost间的exchange，queue相互隔离。就好比一台物理机上(broker)部署多台虚机(vhost)，虚拟机采用不同的用户名密码登录，实现多租户
• Exchange：交换机，消息首先会传递到交换机，由交换机匹配路由键(RoutingKey)决定投递到哪个queue。类比邮政局
• Queue：队列，存储消息的数据结构。类比小区的快递柜
• Binding：绑定，交换机与队列间通过路由键(RoutingKey)进行绑定起来。RoutingKey类比目的地址
• Consumer：消费者，即消息的接受方

安装

1. 下载官方安装文件

2. 执行安装命令

   rpm -ivh erlang-21.3-1.el7.x86_64.rpm
   
   yum install socat -y
   
   rpm -ivh rabbitmq-server-3.8.8-1.el7.noarch.rpm

3. 常用命令

   //开机启动 RabbitMQ 服务
   chkconfig rabbitmq-server on
   
   //启动服务
   文件路径/service rabbitmq-server start
   
   //查看服务状态
   文件路径/service rabbitmq-server status
   
   //停止服务
   文件路径/service rabbitmq-server stop
   
   //开启web管理插件
   rabbit-plugins enable rabbit_managemnet

4. 添加用户

   //创建账号
   rabbitmqctl add_user admin 123
   
   //设置用户角色
   rabbitmqctl set_user_tags admin adminstrator
   
   //设置用户权限
   rabbitmqctl set_permissions -p "/" admin ".*"".*"".*"

---

简单模式

生产者消费者模型

maven项目准备

<dependencies>
    <!-- rabbitmq依赖客户端 -->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.8.0</version>
    </dependency>
    <!-- 操作文件流的依赖 -->
    <dependency>
        <groupId>commons-io</groupId>
        <artifactId>commons-io</artifactId>
        <version>2.6</version>
    </dependency>
</dependencies>

生产者

//生产者：发消息
public class Producer{
    //队列名称
    public static final String QUEUE_NAME = "hello";
    
    //发消息
    public static void main(String[] args){
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //工厂IP 连接RabbitMQ的队列
        factory.setHost("192.168.200.129");
        //用户名、密码
        factory.setUserName("admin");
        factory.setPassword("123");
        
        //创建连接
        Connection connection = factory.newConnection();
        //获取信道
        Channel channel = connection.createChannel();
        /**
        * 创建队列
        * 参数一：队列名称
        * 参数二：是否持久化，默认消息存储在内存中
        * 参数三：该队列是否只供一个消费者进行消费，是否进行消息共享
        * 参数四：是否自动删除
        * 参数五：队列参数
        */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //消息
        String message = "hello world";
        /**
        * 发送消息
        * 参数一：发送到哪个交换机，名称；这里使用的是RabbitMQ默认生成的一个交换机，名称为空串
        * 参数二：路由key（这里用队列名）；这里要注意的是，如果没有将队列和交换机绑定，RabbitMQ会自动的将创建的队列与默认的直接交换机（direct）绑定，路由key就为创建队列的队列名；交换机、路由会在后面提到
        * 参数三：其他参数
        * 参数四：消息体
        */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
    }
}

---

消费者

public class Consumer{
	//队列名称
    public static final String QUEUE_NAME = "hello";
    //接收消息
    public static void main(String[] args){
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //工厂IP 连接RabbitMQ的队列
        factory.setHost("192.168.200.129");
        //用户名、密码
        factory.setUserName("admin");
        factory.setPassword("123");
        
        //创建连接
        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("消费消息被中断");
        }
        /**
        * 接收消息
        * 参数一：队列名
        * 参数二：消费成功后是否要自动应答
        * 参数三：消费者成功接收的回调
        * 参数四：消费者取消消息接收的回调，消费消息被中断时回调
        */
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

---

工作模式

工作队列的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反，我们安排任务在之后执行。我们把任务封装为消息，并将其发送到队列。在后台运行的工作进程将弹出任务，并最终执行作业。当有多个工作线程是，这些工作线程将默认采用轮询的方式一起处理这些任务

轮询分发消息

工具类

public class RabbitMQUtils{
	//得到一个连接的channel
	public static Channel getChannel() throws Exception{
		//创建连接工厂
		ConnectionFactory factory = new ConnectionFactory();
    	//工厂IP 连接RabbitMQ的队列
    	factory.setHost("192.168.200.129");
    	//用户名、密码
    	factory.setUserName("admin");
    	factory.setPassword("123");
        
    	//创建连接
    	Connection connection = factory.newConnection();
     	//获取信道
    	Channel channel = connection.createChannel();
        return channel;
	}
}

工作线程

可以修改回调接口的打印信息来应证采用轮询的方式分发消息

public class Worker01{
    //队列名称
    public static final String QUEUE_NAME = "hello";
    
    //接收消息
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        
         //消费者成功接收的回调
        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println("消费1接收到的消息" + new String(message.getBody()));
        }
        
        //消费消息被中断时回调
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消费1消息被中断");
        }
        //消息接收
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

生产线程

//生产者：发消息
public class Producer{
    //队列名称
    public static final String QUEUE_NAME = "hello";
    
    //发消息
    public static void main(String[] args) throw Exception{
         Channel channel = RabbitMQUtils.getChannel();
        /**
        * 创建队列
        * 参数一：队列名称
        * 参数二：是否持久化，默认消息存储在内存中
        * 参数三：该队列是否只供一个消费者进行消费，是否进行消息共享
        * 参数四：是否自动删除
        * 参数五：队列参数
        */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //从控制台中接收消息
        Scanner scanner = new Scanner(System.in);
        while(scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        }
    }
}

---

不公平分发

虽然默认使用轮询分发，但是在某种场景下这种策略并不是很好，比方说有两个消费者在处理任务，其中一个w1处理任务的速度非常快，而另一个w2处理速度很慢，这个时候如果我们还用轮询分发的方式就会浪费w1的很大部分时间，使其处于空闲状态，而w2会一直在处理事情。

为了避免这种情况，我们可以通过设置参数来取消轮询分发，转为不公平分发

//消费者端控制信道是否为不公平分发
channel.basicQos(1);

意思就是如果这个任务w1还没有处理完或者w1还没有应答 RabbitMQ ， RabbitMQ 先别分配给w1，然后 RabbitMQ 就会把该任务分配给没有那么忙的空闲消费者，当然如果所有的消费者都没有完成手上任务，队列还在不停的添加新任务，队列有可能就会遇到队列被撑满的情况，这个时候就只能添加新的worker或者改变其他存储任务的策略。

---

预取值分发

RabbitMQ的信道上肯定不止只有一个消息，因此这里就存在一个未确认的消息缓冲区，因此希望开发人员能限制此缓冲区的大小，以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用 channel.basicQos() 方法设置“预取计数”值来完成的。该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量，RabbitMQ 将停止在通道上传递更多消息，除非至少有一个未处理的消息被确认。总的来说就是限制一个消费者连接的信道上面能够同时存在多少条数据。

//消费者端控制信道能存放数据量的大小
channel.basicQos(5); //其实只要值大于0都是预取值，但是值为1的时候相当于不公平分发

channel.basicQos(2);

---

消息应答

概念

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分之后突然挂了，会发生什么情况？RabbitMQ一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续发送给该消费者的这条消息，因为它无法接收到。

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

自动应答

消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前，消费者那边出现连接或者channel 关闭，那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制,当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。

---

手动应答

Channel.basicAck() //用于肯定应答，RabbitMQ可以将消息丢弃了
Channel.basicNack() //用于否定应答
Channel.basicReject() //与basicNack()相比少了一个参数，表示不处理该消息，直接拒绝，可以丢弃

批量应答

未开启批量应答时只会提交对当前消息的应答

如果开启批量应答会提交对之前多个消息的应答，但之前的消息可能还未处理完，所以建议不要开启批量应答

消息重新入队

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或TCP连接丢失)，导致消息未发送ACK确认，RabbitMQ将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

消息重新入队保证了消费者宕机不会丢失消息

---

手动应答示例

工作线程

public class Worker01{
    //队列名称
    public static final String QUEUE_NAME = "hello";
    
    //接收消息
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        
         //消费者成功接收的回调
        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println("接收到的消息" + new String(message.getBody()));
            
            //通过线程睡眠的方式模拟一个线程处理快一个线程处理慢，处理过程出错是否能交给另一个线程处理
            SleepUtils.sleep(1);
            //SleepUtils.sleep(30);
            
            //手动应答
            //参数一：消息在RabbitMQ中的一个标记
            //参数二：是否批量应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        }
        
        //消费消息被中断时回调
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消费消息被中断");
        }
        //消息接收
        //关闭参数二的自动应答
        channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);
    }
}

生产线程

//生产者：发消息
public class Producer{
    //队列名称
    public static final String QUEUE_NAME = "hello";
    
    //发消息
    public static void main(String[] args) throw Exception{
        Channel channel = RabbitMQUtils.getChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //从控制台中接收消息
        Scanner scanner = new Scanner(System.in);
        while(scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        }
    }
}

---

持久化

刚刚我们已经看到了如何处理任务不丢失的情况，但是如何保障当RabbitMQ服务停掉以后消息生产者发送过来的消息不丢失呢？默认情况下RabbitMQ退出或由于某种原因崩溃时，它忽视队列和消息，除非告知它不要这样做。确保消息不会丢失需要做两件事:我们需要将队列和消息都标记为持久化。

队列持久化

之前我们创建的队列都是非持久化的，RabbitMQ 如果重启的话，该队列就会被删除掉，如果要队列实现持久化需要在声明队列的时候把第二个参数设置为true；注意的是不能对之前创建的队列进行修改，需要创建一个新的队列，或者先删除这个队列

//设置生产者创建持久化的队列
channel.queueDeclare(QUEUE_NAME,true,false,false,null);

---

消息持久化

想要让消息持久化需要在消息生产者修改代码，在发送消息给队列的时候添加MessageProperties.PERSISTENT_TEXT_PLAIN参数

//设置生产者发送的消息为持久化消息
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉RabbitMQ将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强，所以还需要发布确认来辅助保证持久化。但是对于我们的简单任务队列而言，这已经绰绰有余了。

---

发布确认

生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息陂投递到所有匹配的队列之后，RabbitMQ 就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，RabbitMQ 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 队列也可以设置basic.ack的 multiple域，表示到这个序列号之前的所有消息都已经得到了处理。

confirm模式最大的好处在于他是可以异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack,消息。

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

//生产者创建信道时候开启发布确认模式
channel.confirmSelect();

单个确认

这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布,

channel.waitForConfirmsOrDie()：批量消息确认，它会同时向服务中确认之前当前通道中发送的所有的消息是否已经全部成功写入，这个方法没有任何的返回值，如果服务器中有一条消息没有能够成功或向服务器发送确认时服务不可访问都被认定为消息确认失败，可能有有消息没有发送成功，我们需要进行消费的补发。有参的重载方法，其参数为long类型的超时时间

channel.waitForConfirms()：这个方法只有在消息被确认的时候才返回true

示例：

public static void comfirmMessage(String message) throws Exception{
	Channel channel = RabbitMQUtils.getChannel();
	channel.queueDeclare(QUEUE_NAME,true,false,false,null);
    //开启发布确认
    channel.comfirmSelect();
    
    Scanner scanner = new Scanner(System.in);
    //发送消息
    while(scanner.hasNext()){
        String message = scanner.next();
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        //判断消息是否被确认
        //这样一个个确认队列响应的就是单个发布确认
        if(channel.waitForConfirms()){
            System.out.println("消息发送成功！");
        }
    }
}

批量确认

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

示例：

public static void comfirmMessage(String message) throws Exception{
	Channel channel = RabbitMQUtils.getChannel();
	channel.queueDeclare(QUEUE_NAME,true,false,false,null);
    //开启发布确认
    channel.comfirmSelect();
    //批量确认信息的大小 
    int messagesSize = 100;
    //未确认消息个数
    int uncomfirmMessage = 0;
    
    Scanner scanner = new Scanner(System.in);
    //发送消息
    while(scanner.hasNext()){
        String message = scanner.next();
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        //增加一个未确认的消息
        uncomfirmMessage++;
        //判断消息是否被确认
        //像这样一次性确认多条消息就是批量发布确认
        if(uncomfirmMessage == messagesSize){
            channel.waitForConfirmsOrDie(100);
            System.out.println("批量消息发送成功！");
            uncomfirmMessage = 0; 
        }
    }
}

异步确认

异步确认发布是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功

那如何处理未确认的消息呢？最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用ConcurrentSkipListMap这个跳表在回调函数与发布线程之间进行消息的传递。

示例：

public static void comfirmMessage(String message) throws Exception{
	Channel channel = RabbitMQUtils.getChannel();
	channel.queueDeclare(QUEUE_NAME,true,false,false,null);
    //开启发布确认
    channel.comfirmSelect();
    
    //消息确认成功回调函数
    /*
    * deliveryTag：消息的标记
    * multiple：是否有多条消息被确认，即批量确认，小于等于当前编号的都被确认了；这个参数是RabbitMQ决定的，并不是生产者发送决定的
    * 如果是批量的确认，失败则需要批量的处理
    */
    ConfirmCallback ackCallback = (deliveryTag,multiple)->{
        //如果是批量的
        if(multiple){
            ...
        }else{
            ...
        }
    };
    
    //消息确认失败回调函数
    ConfirmCallback nackCallback = (deliveryTag,multiple)->{
        //如果是批量的
        if(multiple){
            ...
        }else{
            ...
        }
    };
    
    //监听器 监听哪些消息成功了，哪些消息失败了
    //参数一：消息确认成功回调函数
    //参数二：消息确认失败回调函数
    channel.addConfirmListener(ackCallback,nackCallback);
    
    Scanner scanner = new Scanner(System.in);
    //批量发送消息
    while(scanner.hasNext()){
        String message = scanner.next();
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
    }
}

对异步确认进行优化，处理失败的消息

public static void publishMessageAsync() throws Exception {
        Channel channel = RabbitUtils.getChannel();
        channel.queueDeclare(PUBLISH_QUEUE, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        /*
        * 线程安全有序的一个哈希表，适用于高并发情况下
        * 优点1：能很好的将序号（key）和消息（value）关联起来
        * 优点2：轻松批量删除条目，只要给到序号
        * 优点3：适用于高并发（多线程）
        * */
        ConcurrentSkipListMap<Long, String> outstandingConfirms= new ConcurrentSkipListMap<>();
 
        //消息确认成功回调
        ConfirmCallback confirmCallback = (deliveryTag, multiple) ->{
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag);
                confirmed.clear();
            } else {
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("已确认的消息" + deliveryTag);
        };
        //消息确认失败回调
        //deliveryTag :  消息的标记
        //multiple : 是否批量确认
        ConfirmCallback nackCallback = (deliveryTag, multiple)->{
            System.out.println("未确认的消息" + deliveryTag);
            System.out.println("outstandingConfirms = " + outstandingConfirms);
        };
        //开启消息监听，监听哪些消息成功，哪些消息失败
        channel.addConfirmListener(confirmCallback, nackCallback);
 
        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_NUM; i++) {
            String message = i+"";
            channel.basicPublish("", PUBLISH_QUEUE, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
            //outstandingConfirms中存放序号和消息
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
        }
        long end = System.currentTimeMillis();
        System.out.println("异步确认时间 ：end-begin = " + (end - begin));
}

---

交换机

之前所讲的都是简单模式、工作模式，而引入交换机后可以使用 发布/订阅模式

发布/订阅模式可以让消息不止能被消费一次，虽然一个队列中的消息只能被消费一次，但只要把消息发布到不同的队列，就可以实现一个消息被消费多次。

概念

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

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

交换机类型：直接（direct）、主题（topic）、标题（headers）、扇出（fanout）

绑定

绑定就是交换机和队列之间的桥梁，它告诉交换机和哪个队列进行了绑定。比如说下面这张图告诉我们的就是X与Q1和Q2进行了绑定

在网页插件上我们可以创建一个交换机并且跟某一个队列绑定

当绑定了之后我们再使用channel.basicPublish()的时候

• 第一个参数就为交换机的名称
• 第二个参数就应为路由key（Routing key）

channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,null,message);

---

Fanout类型

这种类型非常简单，就是将接收到的所有消息广播到它知道的所有队列中，也就是广播给这个交换机所绑定的所有队列上

消费者

public class Receive{
	//交换机名
	public static final String EXCHANGE_NAME = "exchange";
    
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        //声明一个交换机
        /**
        * 参数一：交换机名
        * 参数二：交换机类型
        */
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
        //声明一个临时队列，队列的名称是随机的，且不是持久化的
        String queueName = channel.queueDeclare().getQueue();
        
        //绑定交换机与队列
        /**
        * 参数一：队列名
        * 参数二：交换机名
        * 参数三：routing key（因为这里使用临时队列，所以 routing key 为空串）
        */
        channel.queueBind(queueName,EXCHANGE_NAME,"");
        
         //消费者成功接收的回调
        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println("接收到的消息" + new String(message.getBody()));
        }
        //消费消息被中断时回调
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消费消息被中断");
        }
        //消息接收 自动应答
        channel.basicConsume(queueName,true,deliverCallback,cancelCallback);
    }
}

生产者

public class Producer{
	//交换机名
	public static final String EXCHANGE_NAME = "exchange";
    
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        
        Scanner scanner = new Scanner(System.in);
        //批量发送消息
        while(scanner.hasNext()){
            String message = scanner.next();
            //参数一：交换机名
            //参数二：routing key（但是这里没有作用，因为fanout类型交换机会广播给所有绑定的队列）
            channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
        }
    }
}

启动一个生产者和两个消费者，之前是由生产者负责创建队列，但是这次换成消费者创建队列和交换机，因为如果是生产者创建队列的话，只能创建一个队列。启动完后生产者发送一条消息给交换机，交换机将广播给所有队列，路由key并不影响广播

---

Direct类型

上述 Fanout 类型的交换机会将所有的消息广播给所有的消费者，对此我们想做一些改变，例如我们希望将消息发送给指定路由key的队列，与我们指定路由key不同的队列接收不到消息，这时候我们就可以使用 Direct 类型的交换机

Direct 交换机还支持多重绑定，可以绑定多个Routing key

消费者

public class Receive{
	//交换机名
	public static final String EXCHANGE_NAME = "exchange";
    //队列名
	public static final String QUEUE_NAME = "queue1";
    
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        //声明一个交换机
        /**
        * 参数一：交换机名
        * 参数二：交换机类型
        */
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
        //声明一个队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        
        //绑定交换机与队列
        /**
        * 参数一：队列名
        * 参数二：交换机名
        * 参数三：指定一个routing key
        */
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"queue1");
        //支持多重绑定，可以绑定多个Routing key
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"queue2");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"queue3");
        
         //消费者成功接收的回调
        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println("接收到的消息" + new String(message.getBody()));
        }
        //消费消息被中断时回调
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消费消息被中断");
        }
        //消息接收 自动应答
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

生产者

public class Producer{
	//交换机名
	public static final String EXCHANGE_NAME = "exchange";
    
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        
        Scanner scanner = new Scanner(System.in);
        //批量发送消息
        while(scanner.hasNext()){
            String message = scanner.next();
            //参数一：交换机名
            //参数二：routing key
            channel.basicPublish(EXCHANGE_NAME,"queue1",null,message.getBytes());
        }
    }
}

启动一个生产者和两个消费者，其中两个消费者所绑定的 Routing key 是不同的，启动完后生产者发送一条指定 Routing key 为 queue1的消息给交换机，交换机将会把这一条消息转发给绑定的 Routing key 为queue1的队列，其他的队列不会收到这条消息

---

Topics类型

发送到类型是 topic 交换机的消息的Routing key不能随意写，必须满足一定的要求，它必须是一个单词列表，以点号分隔开。这些单词可以是任意单词，比如说: “queue1.queue2.queue3” 这洋的。当然这个单词列表最多不能超过255个字节。

同时可以使用替换符

*：可以代替一个单词

#：可以代替零个或多个单词

代码与其他类型交换机演示代码基本相同这里就不代码演示了，就上图的情况我们来分析一下

Routing key	接收情况
test.queue1.test	队列1接收到消息
test.test.queue2	队列2接收到消息
test.queue1.queue2	队列1、队列2接收到消息
test.queue3	队列2接收到消息
test.test.queue3	队列2接收到消息
queue4	不匹配任何 Routing key 消息丢弃

---

死信队列

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

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

死信来源：消息存活时间过期、消息被拒绝、队列达到最大长度…

普通消费者

public class Receive{
	//普通交换机名
	public static final String NORAML_EXCHANGE = "normal_exchange";
    //普通队列名
	public static final String NORMAL_QUEUE = "normal_queue";
    //死信交换机名
	public static final String DEAD_EXCHANGE = "dead_exchange";
    //死信队列名
	public static final String DEAD_QUEUE = "dead_queue";
    
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        
        //声明两个交换机
        channel.exchangeDeclare(NORAML_EXCHANGE,"direct");
        channel.exchangeDeclare(DEAD_EXCHANGE,"direct");
        
        /**声明普通队列*/
        //通过第五个参数设置普通队列额外的参数
        Map<String,Object> args = new HashMap<>();
        //设置消息过期时间（也可以由生产者设置） 单位是豪秒 这里使用消息过期来模拟死信
        // args.put("x-message-ttl",100000);
        //设置队列最大长度（可以用来模拟超出队列长度成为死信） 
        //args.put("x-max-length",6);
        //设置死信交换机
        args.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信路由key
        args.put("x-dead-letter-routing-key","dead_queue");
        //这里的逻辑是因为队列要转发死信到死信交换机，所以需要知道死信交换机名称；且为了发送到死信队列，需要知道死信队列路由key，类似于生产者
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,args);
        
        //声明死信队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        //绑定交换机与队列
        channel.queueBind(NORMAL_QUEUE,NORAML_EXCHANGE,"normal_queue");
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"dead_queue");
        
         //消费者成功接收的回调
        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println("接收到的消息" + new String(message.getBody()));
        }
        //消费消息被中断时回调
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消费消息被中断");
        }
        //消息接收 自动应答
        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);
    }
}

死信消费者

public class Receive{
    //死信交换机名
	public static final String DEAD_EXCHANGE = "dead_exchange";
    //死信队列名
	public static final String DEAD_QUEUE = "dead_queue";
    
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        
        //消费者成功接收的回调
        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println("接收到的死信消息" + new String(message.getBody()));
        }
        //消费消息被中断时回调
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消费消息被中断");
        }
        //消息接收 自动应答
        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);
    }
}

生产者

public class Producer{
	//交换机名
	public static final String EXCHANGE_NAME = "exchange";
    
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        //设置消息过期时间 单位是毫秒
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
        
        Scanner scanner = new Scanner(System.in);
        //批量发送消息
        while(scanner.hasNext()){
            String message = scanner.next();
            //参数一：交换机名
            //参数二：routing key
            channel.basicPublish(EXCHANGE_NAME,"normal_queue",null,message.getBytes());
        }
    }
}

我们先启动普通消费者，再启动生产者和死信消费者。生产者发送消息后，过了10s没有消费者消费，这时我们就能看见死信消费者消费这条消息了，因为消息的过期时间设置为了10s，过了10s没人消费就会转发到死信队列

---

延迟队列

延迟队列最重要的特性就体现在它的延时属性上，演示队列中的元素是希望在指定时间到了以后或之前取出和处理。简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列

其实延迟队列跟死信队列的思想基本是一致的，只不过延迟队列是仅针对消息过期进行处理，就是想让消费者延迟多少秒之后才收到消息

这里使用springboot整合RabbitMQ，且基于死信队列来写代码

配置类

之前我们都是在生产者或者消费者端创建队列和交换机，整合了springboot就不需要了，整合完springboot可以在一个配置类中声明队列、交换机

@Configuration
public class TtlQueueConfig{
    //普通交换机名
	public static final String TTL_EXCHANGE = "ttl_exchange";
    //普通队列名
	public static final String TTL_QUEUE = "ttl_queue";
    //死信交换机名
	public static final String DEAD_EXCHANGE = "dead_exchange";
    //死信队列名
	public static final String DEAD_QUEUE = "dead_queue";
    
    @Bean("ttl_exchange")
    public DirectExchange ttl_exchange(){
        return new DirectExchange(TTL_EXCHANGE);
    }
    
    @Bean("dead_exchange")
    public DirectExchange dead_exchange(){
        return new DirectExchange(DEAD_EXCHANGE);
    }
    
    @Bean("ttl_queue")
    public Queue ttl_queue(){
        Map<String,Object> args = new HashMap<>();
        //设置死信交换机
        args.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信路由key
        args.put("x-dead-letter-routing-key","dead_queue");
        return new QueueBuilder.durable(TTL_QUEUE).withArguments(args).build();
    }
    
    @Bean("dead_queue")
    public Queue dead_queue(){
        return new QueueBuilder.durable(DEAD_QUEUE).build();
    }
    
    @Bean
    public Binding TTLExchangeBindQueue(@Qualifier("ttl_queue") Queue ttl_queue
                                       @Qualifier("ttl_exchange") DirectExchange ttl_exchange){
        return BindingBuilder.bind(ttl_queue).to(ttl_exchange).with("ttl_queue");
    }
    
    public Binding DeadExchangeBindQueue(@Qualifier("dead_queue") Queue dead_queue
                                       @Qualifier("dead_exchange") DirectExchange dead_exchange){
        return BindingBuilder.bind(dead_queue).to(dead_exchange).with("dead_queue");
    }
}

生产者

@RestController
@RequestMapping("/ttl")
public class SendMsgController{
    @Autowired
    private RabbitTemplate rabbitTemplate;
	//发送消息
	@GetMapping("/sendMsg/{message}/{ttlTime}")
	public void sendMsg(@PathVariable String message, @PathVariable String ttlTime){
		rabbitTemplate.convertAndSend("ttl_exchange","ttl_queue",message,msg->{
           	//设置过期时间
            msg.getMessageProperties().setExpiration(ttlTime);
            return msg;
        });
	}
}

消费者

@Component
public class DeadLetterConsumer{
	//接收消息
    @RabbitListener(queues="dead_queue")
    public void receiveDeadLetter(Message message, Channel channel) throws Exception{
        String msg = new String(message.getBody());
        System.out.println(msg);
    }
}

基于死信队列实现延迟队列所存在的问题

由于队列的规则是先进先出的，所以当发了一条过期时间为20s的消息后，再发一条过期时间为2s的消息，过了2s之后这个消息并不会被消费掉，因为RabbitMQ只会检查第一个消息是否过期，如果第一个消息的延迟时间很长，那么第二个消息并不会优先得到执行

基于插件实现延迟队列

上述提到的基于死信队列实现延迟队列所存在的问题可以通过RabbitMQ插件解决

在官网上下载https://www.rabbitmq.com/community-plugins.html，下载**rabbitmq_delayed_message_exchange插件**，然后解压放置到RabbitMQ的插件目录。进入RabbitiMQ的安装目录下的plgins目录，执行下面命令让该插件生效，然后重启RabbitMQ

rabbitmq-plugins enable rabbitmq _delayed_message_exchange

安装好插件之后交换机会多出一种x-delayed-message类型的交换机，我们现在通过交换机来实现延迟队列，而不是通过死信队列来实现

配置类

@Configuration
public class TtlQueueConfig{
    //延迟交换机
	public static final String DELAYED_EXCHANGE = "delayed_exchange";
    //延迟队列名
	public static final String DELAYED_QUEUE = "delayed_queue";
    //延迟路由key
	public static final String DELAYED_ROUTING_KEY = "delayed_queque";
    
    @Bean("delayed_exchange")
    public CustomExchange delayedExchange(){
        Map<String,Object> args = new HashMap<>();
        //设置延迟类型
        args.put("x-delayed-type","direct");
        /**
        * 参数一：交换机名
        * 参数二：交换机类型
        * 参数三：是否需要持久化
        * 参数四：是否需要自动删除
        * 参数五：其他的参数
        */
        return new CustomExchange(DELAYED_EXCHANGE,"x-delayed-message",true,false,args);
    }
    
    @Bean("delayed_queue")
    public Queue delayedQueue(){
        return new QueueBuilder.durable(TTL_QUEUE).build();
    }
    
    @Bean
    public Binding DelayedExchangeBindQueue(@Qualifier("delayed_queue") Queue delayed_queue
                                       @Qualifier("delayed_exchange") DirectExchange delayed_exchange){
        return BindingBuilder.bind(delayed_queue).to(delayed_exchange).with(DELAYED_ROUTING_KEY);
    }
   
}

生产者

@RestController
@RequestMapping("/ttl")
public class SendMsgController{
    @Autowired
    private RabbitTemplate rabbitTemplate;
	//发送消息
	@GetMapping("/sendMsg/{message}/{delayTime}")
	public void sendMsg(@PathVariable String message, @PathVariable String delayTime){
		rabbitTemplate.convertAndSend("delayed_exchange","delayed_queque",message,args->{
           	//设置过期时间
            args.getMessageProperties().setDelay(delayTime);
            return args;
        });
	}
}

消费者

正常消费逻辑即可

---

发布确认——springboot版本

配置文件

spring.rabbitmq.publisher-confirm-type=correlated //针对消息回调开启
spring.rabbitmq.publisher-returns=true //针对消息退回开启

配置类

@Configuration
public class TtlQueueConfig{
    //交换机
	public static final String COMFIRM_EXCHANGE = "comfirm_exchange";
    //队列名
	public static final String COMFIRM_QUEUE = "comfirm_queue";
    //路由key
	public static final String COMFIRM_ROUTING_KEY = "comfirm_queue";
    
    @Bean("comfirm_exchange")
    public DirectExchange comfirmExchange(){
        return new DirectExchange();
    }
    
    @Bean("comfirm_queue")
    public Queue comfirmQueue(){
        return new QueueBuilder.durable(COMFIRM_QUEUE).build();
    }
    
    //绑定
    @Bean
    public Binding DelayedExchangeBindQueue(@Qualifier("comfirm_queue") Queue comfirm_queue
                                       @Qualifier("comfirm_exchange") DirectExchange comfirm_exchange){
        return BindingBuilder.bind(comfirm_queue).to(comfirm_exchange).with(COMFIRM_ROUTING_KEY);
    }
   
}

生产者

@RestController
@RequestMapping("/comfirm")
public class SendMsgController{
    @Autowired
    private RabbitTemplate rabbitTemplate;
	//发送消息
	@GetMapping("/sendMsg/{message}")
	public void sendMsg(@PathVariable String message){
        // correlationData 保存消息的ID及相关信息
        CorrelationData correlationData = new CorrelationData("1");
		rabbitTemplate.convertAndSend("comfirm_exchange","comfirm_queue",message,correlationData);
	}
}

回调接口

这里与上面的发布确认有点区别的是，这里的回调借口不会等队列将数据持久化之后才调用这个接口，而是到达交换机就调用这个接口

confirm()：处理消息能否到达交换机

returnedMessage()：处理消息能否到达队列

@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback{
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @PostConstruct
    public void init(){
        //注入 
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }
    
	/**确认回调方法 到达交换机就触发
	* 参数一：correlationData 保存回调消息的ID及相关信息
	* 参数二：交换是否机收到消息
	* 参数三：原因
    */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause){
        if(ack){
            System.out.println("交换机收到消息");
        }else{
            System.out.println("交换机未收到消息，由于：" + cause);
        }
    }
    
    /**回退消息	可以在消息传递过程中不可达目的地时将消息回退给生产者
    * 参数一：封装后的消息
    * 参数二：回退code
    * 参数三：回退原因
    * 参数四：回退的交换机
    * 参数五：发送方要发送的队列的路由key
    */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey){
        System.out.println("消息：" + new String(message.getBody()) + "被交换机：" + exchange + "回退，原因是：" + replyText);
    }
}

消费者

@Component
public class DeadLetterConsumer{
	//接收消息
    @RabbitListener(queues="comfirm_queue")
    public void receiveDeadLetter(Message message, Channel channel) throws Exception{
        String msg = new String(message.getBody());
        System.out.println(msg);
    }
}

---

回退优化，对回退的消息进行处理（备份和警告）

有时候我们并不想仅仅把消息回退给生产者，而是想做其他的操作，这时候就可以使用备份交换机

普通交换机

alternate-exchange参数用于绑定备份交换机

备份交换机

应设置为Fanout类型交换机这样可以广播到每个绑定的队列

注意：当我们同时开启消息回退方式和备份交换机方式时，备份交换机的优先级更高

---

幂等性

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

消息重复消费

消费者在消费MQ中的消息时，MQ已把消息发送给消费者，消费者在给MQ返回ack时网络中断，故MQ未收到确认信息，该条消息会重新发给其他的消费者，或者在网络重连后再次发送给该消费者，但实际上该消费者已成功消费了该条消息，造成消费者消费了重复的消息。

解决思路

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

• 唯一ID+指纹码机制

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

• Redis原子性

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

---

优先级队列

使用场景

在我们系统中有一个订单催付的场景，我们的客户催付之后会给消费者发送短信，而有些客户可能是vip，有些客户是超级vip，这时候我们发短信的优先级就需要改变，应该给优先级高的客户先发，这里就可以使用到优先级队列

使用优先级队列时候，队列会先对消息的优先级进行排序

方式一：可以通过web端添加

方式二：代码添加

配置类

@Bean("priority_queue")
public Queue priorityQueue(){
    return new QueueBuilder.durable(COMFIRM_QUEUE)
        .withArgument("x-max-priority", 10)
        .build();
}

生产者

rabbitTemplate.convertAndSend("exchange_priority", "priority", message+i, args-> {
    //设置优先级
    args.getMessageProperties().setPriority(5);
    return args;
});

---

惰性队列

RabbitMQ从3.6.0版本开始引入了惰性队列的概念。惰性队列会尽可能的将消息存入磁盘中，而在消费者消费到相应的消息时才会被加载到内存中，它的一个重要的设计目标是能够支持更长的队列，即支持更多的消息存储。当消费者由于各种各样的原因(比如消费者下线、宕机亦或者是由于维护而关闭等)而致使长时间内不能消费消息造成堆积时，惰性队列就很有必要了。

默认情况下，当生产者将消息发送到 RabbitMQ 的时候，队列中的消息会尽可能的存储在内存之中，这样可以更加快速的将消息发送给消费者。即使是持久化的消息，在被写入磁盘的同时也会在内存中驻留一份备份。当 RabbitMQ 需要释放内存的时候，会将内存中的消息换页至磁盘中，这个操作会耗费较长的时间，也会阻塞队列的操作，进而无法接收新的消息。虽然RabbitMQ的开发者们一直在升级相关的算法，但是效果始终不太理想，尤其是在消息量特别大的时候。

队列具备两种模式：default 和 lazy。默认的为 default 模式，在 3.6.0 之前的版本无需做任何变更。lazy模式即为惰性队列的模式，可以通过声明队列的时候在参数中设置，也可以通过Policy 的方式设置，如果一个队列同时使用这两种方式设置的话，那么 Policy 的方式具备更高的优先级。

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

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

内存开销对比

在发送 1 百万条消息，每条消息大概占 1KB 的情况下，普通队列占用内存是 1.2GB，而惰性队列仅仅占用 1.5MB ，内存中剩下的是一些索引，方便查找。

---

RabbitMQ集群

搭建集群

1. 准备3台安装了RabbitMQ机器

   主机名	ip
   node1	10.211.55.74
   node2	10.211.55.75
   node3	10.211.55.76

2. 配置各个节点的host文件，让各个节点都能互相识别对方

   vim /etc/hosts
   10.211.55.74 node1
   10.211.55.75 node2
   10.211.55.76 node3

3. 确保每个节点的 cookie 文件使用的是同一个值

   ##使用远程复制命令，将第一台的cookie文件复制给其他两台
   ##node1
   scp /var/lib/rabbitmq/.erlang.cookie root@node2:/var/lib/rabbitmq/.erlang.cookie
   scp /var/lib/rabbitmq/.erlang.cookie root@node3:/var/lib/rabbitmq/.erlang.cookie

4. 重启 RabbitMQ 服务和Erlang虚拟机，三台机器上分别执行以下命令

   rabbitmq-server-detached

5. 在各个节点执行

   ##node2
   rabbitmqctl stop_app
   rabbitmqctl reset
   rabbitmqctl join_cluster rabbit@node1
   rabbitmqctl start_app
   
   ##node3
   rabbitmqctl stop_app
   rabbitmqctl reset
   rabbitmqctl join_cluster rabbit@node1
   rabbitmqctl start_app

6. 集群状态

   rabbitmq cluster_status

7. 需要重新设置用户

   //创建账号
   rabbitmqctl add_user admin 123
   
   //设置用户角色
   rabbitmqctl set_user_tags admin adminstrator
   
   //设置用户权限
   rabbitmqctl set_permissions -p "/" admin ".*"".*"".*"

---

镜像队列

使用镜像队列的原因

假设 RabbitMQ 集群中只有一个节点，那么该节点的失效将导致整体服务的临时不可用。即访问不到在这个节点下创建的队列。

引入镜像队列的机制，可以将队列镜像到集群中的其他节点上，如果集群中的一个节点失效了，队列能自动地切换到镜像中的另一个节点上以保证服务的可用性。

web插件 —— Admin —— Policies

备份了几份会自动分配给集群上的其他节点，主一份，其他 n 份

当节点1宕机了，其备份的队列会转到其他的节点上，总的来说就是在存活的节点上一定保持 n 份数据

---

高可用负载均衡

上述的操作都还是只连接了一台 RabbitMQ 服务器进行操作，即连接工厂只连接了一台 RabbitMQ 的 ip factory.setHost("192.168.200.129")，这样写死 ip 对于集群来说这样是体现不出集群的优点的

我们可以通过nginx + keepalived来实现负载均衡

这里涉及到其他技术，就不再赘述

---

跨集群

现在大型的网站系统，为了实现异地容灾，一般采用多机房或者混合云部署。我们来讨论下单个集群在这种场景下的集中方案

1、仅部署一个机房

单个集群仅部署在北京机房，南京机房的发布者和消费者跨机房访问。这个方案有以下几个问题。

1. 无法做到容灾，一旦北京机房挂了，整个集群不可用
2. 降低吞吐量，南京机房的客户端发送请求后，会阻塞住，直到节点回复确认，由于两个机房的存在延迟(假设北京到南京机房的RTT在30ms)，导致每次请求时间增加，降低了吞吐量

2、延展机房部署

单个集群跨机房部署，节点1，节点2部署到北京机房，节点3部署到南京机房。这个方案通过镜像队列在节点1，节点2与节点3间互备，能做到部分的容灾。但是也有如下问题：

1. 跨机房请求，部分消费者和发布者还是需要跨机房请求，与方案1类似，导致请求时间增加，降低了吞吐量
2. 脑裂，异地间的网络环境复杂，网络波动会导致分区，进而”脑裂”，单个集群是无法做到跨可用区的

​ 单个集群是无法满足跨机房场景，需要采用多集群部署解决这个问题，跨集群间的”桥接”可以通过Federation，Shovel两种方式。

Federation

Federation交换机

为什么使用Federation交换机

（broker北京），（broker深圳）彼此之间相距甚远，网络延迟是一个不得不面对的问题。有一个在北京的业务（Client北京）需要连接（broker北京），向其中的交换器exchangeA发送消息，此时的网络延迟很小，（Client北京）可以迅速将消息发送至exchangeA中，就算在开启了publisherconfirm机制或者事务机制的情况下，也可以迅速收到确认信息。

此时又有个在深圳的业务（Client深圳）需要向exchangeA发送消息，那么（Client深圳）（broker北京）之间有很大的网络延迟，（Client深圳）将发送消息至exchangeA会经历一定的延迟，尤其是在开启了publisherconfirm机制或者事务机制的情况下，（Client深圳）会等待很长的延迟时间来接收（broker北京）的确认信息，进而必然造成这条发送线程的性能降低，甚至造成一定程度上的阻塞。

这里我们应该让（Client北京）都去连接（broker北京），（Client深圳）都去连接（broker深圳），但是（broker北京），（broker深圳）之间的数据如何同步和一致呢，这里我们使用 Federation 插件就能很好地解决这个问题

在rabbitmq的分布式集群中，我们都是通过配置集群的模式进行分布式部署的，一般都是在内网中使用客户端进行连接调用，但是如果我们遇到大型的分布式集群的时候，比如一个部署在南方，一个部署在北方，然而rabbitmq集群只是部署在了南方，如果北方的分布式程序要来调用rabbitmq集群，那么就只能通过网络来进行远程调用了，在这个过程中我们是不能保证网络的状态的，因此rabbitmq也考虑到了这个因素，因此也就有了federation插件的诞生，它主要解决了以下两个问题：

1.针对不同的erlang版本和rabbitmq版本，只要都是采用的AMQP 0.9.1作为传输协议都可以进行连接，而不需要建立集群

2.针对广域网中的复杂网络环境，针对不在同一个地区的分布式部署，可以采用federation联合的方式进行数据传输。它也可以在同一台服务器的不同virtual上面进行数交互

但是我们需要注意的是federation联合的数据在queue中并没有被转移到联合的一方，而是仍然保留在联合的一方

搭建

1. 在每台机器上开启 Federation 相关插件，并准备好两个交换机和两个队列

   rabbitmq-plugins enable rabbitmq_federation
   rabbitmq-plugins enable rabbitmq_federation_management

2. 进行配置；web插件 —— Admin —— Federation Upstreams

   

   下面我们依次解释下federation中的参数信息：

   • name: 新增的federation联合插件的上游名称，这个可以随便取名

   • uri：上游federation联合rabbitmq的地址，上游是指订阅的服务器节点，它的定义方式有4种：

   1. amqp://server-name

      非SSL方式直接连接到上游服务器，不需要验证，如amqp://192.168.0.105

   2. amqp://user:password@server-name/my-vhost

      通过用户名和密码的非SSL方式连接到服务器，并且指定相应的virtual host，需要注意的是如果使用默认的virtual host /，那么IP地址后面什么都不需要写，如amqp://admin:admin@192.168.0.105，如果写成了amqp://admin:admin@192.168.0.105/，那么将会无法进行连接启动

   3. amqps://user:password@server-name?cacertfile=/path/to/cacert.pem&certfile=/path/to/cert.pem&keyfile=/path/to/key.pem&verify=verify_peer

      通过用户名和密码+证书的方式进行SSL连接

   4. amqps://server-name?cacertfile=/path/to/cacert.pem&certfile=/path/to/cert.pem&keyfile=/path/to/key.pem&verify=verify_peer&fail_if_no_peer_cert=true&auth_mechanism=external

      不使用用户名和密码，但需要证书的方式进行SSL连接。

   • prefect count：一次性从上游rabbitmq服务器预期数据的最大数量，默认是1000

   • reconnect delay：网络连接失败后下次重连的等待时间，默认是5秒

   • Acknowledgement Mode：消息确认方式，on-confirm、on-publish和no-ack

   • Trust User-ID：是否信任从上游服务器传来的用户id，默认是yes，设置成no，将会清空从上游服务器传来的user id信息

   接下来是专门提供给federation exchange交换器的参数：

   • Exchange：定义上游服务器的联合的exchange名称，默认情况下的取名与联合的exchange名称相同

   • max hops：消息在被放弃或者说被消费前消息可以传递的最大的联合federation 连接长度，默认是1，一般情况连接长度等于联合的节点数量-1

   • Expires：上游服务器节点保持节点信息的最长时间，单位毫秒，默认的是永久保存

   • Message TTL：在上游节点中消息未被传递时可以保存的时长，单位毫秒，默认是永久保存

   • HA Policy：检查一个联合exchange的上游queue中的x-ha-policy，用于确认该queue是否是一个HA的queue，默认是none表示不是一个HA的queue

3. 添加策略；匹配规则不是以节点作为目标，而是以交换机作为目标；web插件 —— Admin —— Policies

   

• 在上面的参数中name是可以随便取名字
• pattern表示匹配的表达式，用法是正则表达式
• apply to表示应用在exchange还是queue上面，亦或两者都使用
• priority表示优先级，值越大，优先级越高
• definition用于定义使用的配置，这里我们定义的是federation联合，它有federation upstream set和 federation upstream两种方式，set表示集合，定义需要该策略的所有上游名称，一般我们都取值为all，没有set的表示单个定义上游名称

---

Federation队列

联邦队列可以看做互为扩展队列，如图中的Queue1与Queue3建立联邦队列，当Queue3有消息堆积，消费者1优先将Queue3的消费完，此时则会从Queue1拉取消息；反之亦然，如果Queue1的队列消费完成，将会从Queue3中拉取消息，消费者1继续消息，所以Queue1与Queue3间的消息是可以”漂移”的。联邦队列会让消费能力强的一侧多消费些，确保队列间的平衡。

但是这样可能会存在消费者重复消费同一个消息的问题，这时候可以使用上文讲到的，解决消费者重复消费的方法进行解决

联邦交换机的消息流向是单向的，联邦队列消息流向是双向

配置与Federation交换机类似，不再赘述

---

Shovel

以broker1的Queue为源，broker2的Exchange为目标，建立shovel link，发布者发送的Msg1消息，存入queue，Shovel消费后存入broker2的queue。当broker1的queue产生消息堆积时，通过shovel link转移消息到其他集群进行消息，减少broke1的压力。

配置上网查，懒得写 ( ‘▽′)Ψ

---

SpringBoot整合RabbitMQ

引入依赖

<!-- RabbitMQ依赖 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- RabbitMQ测试依赖 -->
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit-test</artifactId>
    <version>2.4.5</version>
</dependency>

修改配置文件

spring.rabbitmq.host=182.92.234.71
spring.rabbitmq.port=8081
spring.rabbitmq.username=admin
spring.rabbitmq.password=123