学习目标
1. 消息中间件概述
2. 安装及配置RabbitMQ
3. RabbitMQ入门
4. RabbitMQ工作模式
5. Spring 整合RabbitMQ
- 5.1. 搭建生产者工程
- 5.2. 搭建消费者工程
6. Spring Boot整合RabbitMQ
RabbitMQ 高级特性
RabbitMQ应用问题
- 消息可靠性保障
- 消息幂等性处理
RabbitMQ集群搭建
- 集群方案的原理

学习目标

能够说出什么是消息中间件
能够安装RabbitMQ
能够编写RabbitMQ的入门程序
能够说出RabbitMQ的5种模式特征
能够使用Spring整合RabbitMQ

1. 消息中间件概述

1.1 什么是消息中间件
MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进
行通信。
MQ，消息队列，存储消息的中间件
向MQ中发送消息的一方为消息的生产者 , 从MQ中获取消息的一方为消费者

1.2 MQ 的优势和劣势
优势: 劣势：
应用解耦 * 系统可用性降低
异步提速 * 系统复杂度提高
削峰填谷 * 一致性问题

1.2.1 应用解耦

使用 MQ 使得应用间解耦，提升容错性和可维护性。

1.2.2 异步提速

用户点击完下单按钮后，只需等待25ms就能得到下单响应 (20 + 5 = 25ms)。
提升用户体验和系统吞吐量（单位时间内处理请求的数目）。

1.2.3 削峰填谷

使用了 MQ 之后，限制消费消息的速度为1000，这样一来，高峰期产生的数据势必会被积压在 MQ 中，高峰
就被“削”掉了.

但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000，直
到消费完积压的消息，这就叫做“填谷”
使用MQ后，可以提高系统稳定性。

1.2.4 小结
1. 应用解耦：提高系统容错性和可维护性
2. 异步提速：提升用户体验和系统吞吐量
3. 削峰填谷：提高系统稳定性

1.3 MQ 的劣势
系统可用性降低

系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

系统复杂度提高

MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何
保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

一致性问题

A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理
失败。如何保证消息数据处理的一致性？
小结 :
既然 MQ 有优势也有劣势，那么使用 MQ 需要满足什么条件呢？
① 生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明
明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。
② 容许短暂的不一致性。
③ 确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本。

1.4 消息队列产品

市场上常见的消息队列有如下：

ActiveMQ：基于JMS
ZeroMQ：基于C语言开发
RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
RocketMQ：基于JMS，阿里巴巴产品
Kafka：类似MQ的产品；分布式消息系统，高吞吐量

1.5 RabbitMQ
RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信中间件，消息队列在分布式系统开发中应用非常广泛。
RabbitMQ官方地址：http://www.rabbitmq.com/
RabbitMQ 基础架构如下图：

RabbitMQ提供了6种模式：简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式（远程调用，不太算MQ；暂不作介绍）；
官网对应模式介绍：https://www.rabbitmq.com/getstarted.html

1.7 AMQP 和 JMS
MQ是消息通信的模型；实现MQ的大致有两种主流方式：AMQP、JMS。
1.7.1. AMQP
AMQP 一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。
RabbitMQ 中的相关概念：

概念	说明
连接Connection	publisher／consumer 和 broker 之间的 TCP 连接
管道Channel	通过连接对象创建 , 用于消息的传输
交换机Exchange	发送的消息首先到达交换机, 与队列绑定. 常用的类型有fanout , direct , topic
队列Queue	消息存储的数据结构

1.7.2. JMS

JMS即Java消息服务（JavaMessage Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

2. 安装及配置RabbitMQ

详细查看 资料/软件/安装RabbitMQ.md 文档。

rabbitmq-plugins enable rabbitmq_management

3. RabbitMQ入门

3.1. 搭建示例工程

3.1.1. 创建工程

3.1.2. 添加依赖

往heima-rabbitmq的pom.xml文件中添加如下依赖：

        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>

3.2. 编写生产者

编写消息生产者com.itheima.producer

package com.itheima.producer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * 发送消息
 */
public class Producer_HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();
        //5. 创建队列Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        参数：
            1. queue：队列名称
            2. durable:是否持久化，当mq重启之后，还在
            3. exclusive：
                * 是否独占。只能有一个消费者监听这队列
                * 当Connection关闭时，是否删除队列
                *
            4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
            5. arguments：参数。
         */
        //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
        channel.queueDeclare("hello_world",true,false,false,null);
        /*
        basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
        参数：
            1. exchange：交换机名称。简单模式下交换机会使用默认的 ""
            2. routingKey：路由名称
            3. props：配置信息
            4. body：发送消息数据
         */
        String body = "hello rabbitmq~~~";
        //6. 发送消息
        channel.basicPublish("","hello_world",null,body.getBytes());
        //7.释放资源
      channel.close();
      connection.close();
    }
}

在执行上述的消息发送之后；可以登录rabbitMQ的管理控制台，可以发现队列和其消息

3.3. 编写消费者

package com.itheima.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer_HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();
        //5. 创建队列Queue
        /*
        参数：
            1. queue：队列名称
            2. durable:是否持久化，当mq重启之后，还在
            3. exclusive：
                * 是否独占。只能有一个消费者监听这队列
                * 当Connection关闭时，是否删除队列
                *
            4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
            5. arguments：参数。
         */
        //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
        channel.queueDeclare("hello_world", true, false, false, null);
        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1. queue：队列名称
            2. autoAck：是否自动确认
            3. callback：回调对象
         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel) {
            /*
                回调方法，当收到消息后，会自动执行该方法
                1. consumerTag：标识
                2. envelope：获取一些信息，交换机，路由key...
                3. properties:配置信息
                4. body：数据
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumerTag：" + consumerTag);
                System.out.println("Exchange：" + envelope.getExchange());
                System.out.println("RoutingKey：" + envelope.getRoutingKey());
                System.out.println("properties：" + properties);
                System.out.println("body：" + new String(body));
            }
        };
        channel.basicConsume("hello_world", true, consumer);
        //关闭资源？不要
    }
}

3.4. 小结

上述的入门案例中中其实使用的是如下的简单模式：

在上图的模型中，有以下概念：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。

queue：消息队列，图中红色部分。生产者向其中投递消息，消费者从其中取出消息。

4. RabbitMQ工作模式

4.1. Work queues工作队列模式

4.1.1. 模式说明

Work Queues与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
特点：一个生产者发送消息完成后，存在多个消费者消费，消费者之间是存在竞争关系.
交换机：使用了默认交换机
routingKey：默认是队列的名称
应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

4.1.2. 代码

Work Queues与入门程序的简单模式的代码是几乎一样的；可以完全复制，并复制多一个消费者进行多个消费者同时消费消息的测试。

1）生产者

package com.itheima.producer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 发送消息
*/
public class Producer_WorkQueues {
  public static void main(String[] args) throws IOException, TimeoutException {
      //1.创建连接工厂
      ConnectionFactory factory = new ConnectionFactory();
      //2. 设置参数
      factory.setHost("172.16.98.133");//ip  默认值 localhost
      factory.setPort(5672); //端口  默认值 5672
      factory.setVirtualHost("/itcast");//虚拟机 默认值/
      factory.setUsername("heima");//用户名 默认 guest
      factory.setPassword("heima");//密码 默认值 guest
      //3. 创建连接 Connection
      Connection connection = factory.newConnection();
      //4. 创建Channel
      Channel channel = connection.createChannel();
      //5. 创建队列Queue
      /*
      queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
      参数：
          1. queue：队列名称
          2. durable:是否持久化，当mq重启之后，还在
          3. exclusive：
              * 是否独占。只能有一个消费者监听这队列
              * 当Connection关闭时，是否删除队列
              *
          4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
          5. arguments：参数。
       */
      //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
      channel.queueDeclare("work_queues",true,false,false,null);
      /*
      basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
      参数：
          1. exchange：交换机名称。简单模式下交换机会使用默认的 ""
          2. routingKey：路由名称
          3. props：配置信息
          4. body：发送消息数据
       */
      for (int i = 1; i <= 10; i++) {
          String body = i+"hello rabbitmq~~~";
          //6. 发送消息
          channel.basicPublish("","work_queues",null,body.getBytes());
      }
      //7.释放资源
    channel.close();
    connection.close();
  }
}

2）消费者1

package com.itheima.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer_WorkQueues1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();
        //5. 创建队列Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        参数：
            1. queue：队列名称
            2. durable:是否持久化，当mq重启之后，还在
            3. exclusive：
                * 是否独占。只能有一个消费者监听这队列
                * 当Connection关闭时，是否删除队列
                *
            4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
            5. arguments：参数。
         */
        //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
        channel.queueDeclare("work_queues",true,false,false,null);
        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1. queue：队列名称
            2. autoAck：是否自动确认
            3. callback：回调对象
         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回调方法，当收到消息后，会自动执行该方法
                1. consumerTag：标识
                2. envelope：获取一些信息，交换机，路由key...
                3. properties:配置信息
                4. body：数据
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              /*  System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);*/
                System.out.println("body："+new String(body));
            }
        };
        channel.basicConsume("work_queues",true,consumer);
        //关闭资源？不要
    }
}

3）消费者2

package com.itheima.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer_WorkQueues2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();
        //5. 创建队列Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        参数：
            1. queue：队列名称
            2. durable:是否持久化，当mq重启之后，还在
            3. exclusive：
                * 是否独占。只能有一个消费者监听这队列
                * 当Connection关闭时，是否删除队列
                *
            4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
            5. arguments：参数。
         */
        //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
        channel.queueDeclare("work_queues",true,false,false,null);
        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1. queue：队列名称
            2. autoAck：是否自动确认
            3. callback：回调对象
         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回调方法，当收到消息后，会自动执行该方法
                1. consumerTag：标识
                2. envelope：获取一些信息，交换机，路由key...
                3. properties:配置信息
                4. body：数据
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              /*  System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);*/
                System.out.println("body："+new String(body));
            }
        };
        channel.basicConsume("work_queues",true,consumer);
        //关闭资源？不要
    }
}

4.1.3. 小结

在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。
工作队列模式总结：

如果生产者发了一个消息，消费者如果是多个的情况下，有且只有一个消费者能消费消息
如果生产者发的是多条消息，各个消费者之间是竞争消费关系。

4.2. 订阅模式类型
订阅模式示例图：

前面2个案例中，只有3个角色：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分

而在订阅模型中，多了一个exchange角色，而且过程略有变化：

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
C：消费者，消息的接受者，会一直等待消息到来。
Queue：消息队列，接收消息、缓存消息。
Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：
- Fanout：广播，将消息交给所有绑定到交换机的队列
- Direct：定向，把消息交给符合指定routing key 的队列
- Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

4.3. Publish/Subscribe发布与订阅模式

4.3.1. 模式说明

1、声明交换机，并且交换机的类型为 fanout
2、发布订阅的工作模式fanout，设置 routingkey=“” 即可
特点：发布订阅是一种广播的方式，只要是订阅了当前的队列，就会全部的接收到消息

4.3.2. 代码

1）生产者

package com.itheima.producer;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * 发送消息
 */
public class Producer_PubSub {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("172.16.98.133");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();

       /*
       参数：
        1. exchange:交换机名称
        2. type:交换机类型
            DIRECT("direct"),：定向
            FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
            TOPIC("topic"),通配符的方式
            HEADERS("headers");参数匹配
        3. durable:是否持久化
        4. autoDelete:自动删除
        5. internal：内部使用。 一般false
        6. arguments：参数
        */
       String exchangeName = "test_fanout";
        //5. 创建交换机
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
        //6. 创建队列
        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);
        //7. 绑定队列和交换机
        /*
        queueBind(String queue, String exchange, String routingKey)
        参数：
            1. queue：队列名称
            2. exchange：交换机名称
            3. routingKey：路由键，绑定规则
                如果交换机的类型为fanout ，routingKey设置为""
         */
        channel.queueBind(queue1Name,exchangeName,"");
        channel.queueBind(queue2Name,exchangeName,"");
        String body = "日志信息：张三调用了findAll方法...日志级别：info...";
        //8. 发送消息
        channel.basicPublish(exchangeName,"",null,body.getBytes());
        //9. 释放资源
        channel.close();
        connection.close();
    }
}

2）消费者1

package com.itheima.rabbitmq.ps;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer1 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);
        //队列绑定交换机
        channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHAGE, "");
        //创建消费者；并设置消息处理
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            /**
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为：" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为：" + envelope.getExchange());
                //消息id
                System.out.println("消息id为：" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer);
    }
}

3）消费者2

package com.itheima.rabbitmq.ps;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer2 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null);
        //队列绑定交换机
        channel.queueBind(Producer.FANOUT_QUEUE_2, Producer.FANOUT_EXCHAGE, "");
        //创建消费者；并设置消息处理
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            /**
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为：" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为：" + envelope.getExchange());
                //消息id
                System.out.println("消息id为：" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));
            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer);
    }
}

4.3.3. 小结

交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。
发布订阅模式与工作队列模式的区别?
1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。
2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。
3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机。

4.4. Routing路由模式

4.4.1. 模式说明

路由模式特点：

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。
Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断5头发不不不3，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

图解：

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息

C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

4.4.2. 代码

在编码上与 Publish/Subscribe发布与订阅模式 的区别是交换机的类型为：Direct，还有队列绑定交换机的时候需要指定routing key。

1）生产者

package com.itheima.producer;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 发送消息
*/
public class Producer_Routing {
  public static void main(String[] args) throws IOException, TimeoutException {
      //1.创建连接工厂
      ConnectionFactory factory = new ConnectionFactory();
      //2. 设置参数
      factory.setHost("172.16.98.133");//ip  默认值 localhost
      factory.setPort(5672); //端口  默认值 5672
      factory.setVirtualHost("/itcast");//虚拟机 默认值/
      factory.setUsername("heima");//用户名 默认 guest
      factory.setPassword("heima");//密码 默认值 guest
      //3. 创建连接 Connection
      Connection connection = factory.newConnection();
      //4. 创建Channel
      Channel channel = connection.createChannel();
     /*
     参数：
      1. exchange:交换机名称
      2. type:交换机类型
          DIRECT("direct"),：定向
          FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
          TOPIC("topic"),通配符的方式
          HEADERS("headers");参数匹配
      3. durable:是否持久化
      4. autoDelete:自动删除
      5. internal：内部使用。 一般false
      6. arguments：参数
      */
     String exchangeName = "test_direct";
      //5. 创建交换机
      channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);
      //6. 创建队列
      String queue1Name = "test_direct_queue1";
      String queue2Name = "test_direct_queue2";
      channel.queueDeclare(queue1Name,true,false,false,null);
      channel.queueDeclare(queue2Name,true,false,false,null);
      //7. 绑定队列和交换机
      /*
      queueBind(String queue, String exchange, String routingKey)
      参数：
          1. queue：队列名称
          2. exchange：交换机名称
          3. routingKey：路由键，绑定规则
              如果交换机的类型为fanout ，routingKey设置为""
       */
      //队列1绑定 error
      channel.queueBind(queue1Name,exchangeName,"error");
      //队列2绑定 info  error  warning
      channel.queueBind(queue2Name,exchangeName,"info");
      channel.queueBind(queue2Name,exchangeName,"error");
      channel.queueBind(queue2Name,exchangeName,"warning");
      String body = "日志信息：张三调用了delete方法...出错误了。。。日志级别：error...";
      //8. 发送消息
      channel.basicPublish(exchangeName,"warning",null,body.getBytes());
      //9. 释放资源
      channel.close();
      connection.close();
  }
}

2）消费者1

package com.itheima.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer_Routing1 {
  public static void main(String[] args) throws IOException, TimeoutException {
      //1.创建连接工厂
      ConnectionFactory factory = new ConnectionFactory();
      //2. 设置参数
      factory.setHost("172.16.98.133");//ip  默认值 localhost
      factory.setPort(5672); //端口  默认值 5672
      factory.setVirtualHost("/itcast");//虚拟机 默认值/
      factory.setUsername("heima");//用户名 默认 guest
      factory.setPassword("heima");//密码 默认值 guest
      //3. 创建连接 Connection
      Connection connection = factory.newConnection();
      //4. 创建Channel
      Channel channel = connection.createChannel();
      String queue1Name = "test_direct_queue1";
      String queue2Name = "test_direct_queue2";
      // 接收消息
      Consumer consumer = new DefaultConsumer(channel){
          /*
              回调方法，当收到消息后，会自动执行该方法
              1. consumerTag：标识
              2. envelope：获取一些信息，交换机，路由key...
              3. properties:配置信息
              4. body：数据
           */
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            /*  System.out.println("consumerTag："+consumerTag);
              System.out.println("Exchange："+envelope.getExchange());
              System.out.println("RoutingKey："+envelope.getRoutingKey());
              System.out.println("properties："+properties);*/
              System.out.println("body："+new String(body));
              System.out.println("将日志信息打印到控制台.....");
          }
      };
         /*
      basicConsume(String queue, boolean autoAck, Consumer callback)
      参数：
          1. queue：队列名称
          2. autoAck：是否自动确认
          3. callback：回调对象
       */
      channel.basicConsume(queue2Name,true,consumer);
      //关闭资源？不要
  }
}

3）消费者2

package com.itheima.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer_Routing2 {
  public static void main(String[] args) throws IOException, TimeoutException {
      //1.创建连接工厂
      ConnectionFactory factory = new ConnectionFactory();
      //2. 设置参数
      factory.setHost("172.16.98.133");//ip  默认值 localhost
      factory.setPort(5672); //端口  默认值 5672
      factory.setVirtualHost("/itcast");//虚拟机 默认值/
      factory.setUsername("heima");//用户名 默认 guest
      factory.setPassword("heima");//密码 默认值 guest
      //3. 创建连接 Connection
      Connection connection = factory.newConnection();
      //4. 创建Channel
      Channel channel = connection.createChannel();
      String queue1Name = "test_direct_queue1";
      String queue2Name = "test_direct_queue2";
      // 接收消息
      Consumer consumer = new DefaultConsumer(channel){
          /*
              回调方法，当收到消息后，会自动执行该方法
              1. consumerTag：标识
              2. envelope：获取一些信息，交换机，路由key...
              3. properties:配置信息
              4. body：数据
           */
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            /*  System.out.println("consumerTag："+consumerTag);
              System.out.println("Exchange："+envelope.getExchange());
              System.out.println("RoutingKey："+envelope.getRoutingKey());
              System.out.println("properties："+properties);*/
              System.out.println("body："+new String(body));
              System.out.println("将日志信息存储到数据库.....");
          }
      };

      /*
      basicConsume(String queue, boolean autoAck, Consumer callback)
      参数：
          1. queue：队列名称
          2. autoAck：是否自动确认
          3. callback：回调对象
       */
      channel.basicConsume(queue1Name,true,consumer);
      //关闭资源？不要
  }
}

4.4.3. 小结

Routing模式要求队列在绑定交换机时要指定routing key，消息会转发到符合routing key的队列。

总结： 1、声明交换机，并且设置交换机的类型为 Direct 2、声明队列，需要将队列绑定到交换机上并且要设置routingkey，交换机会根据生产者发消息的时候指导的routingkey，将消息转发到对应的队列中

4.5. Topics通配符模式

4.5.1. 模式说明

Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！
Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
通配符规则：
#：匹配一个或多个词
*：匹配1个词
举例：
item.#：能够匹配item.insert.abc 或者 item.insert
item.*：只能匹配item.insert

图解：

红色Queue：绑定的是usa.# ，因此凡是以 usa.开头的routing key 都会被匹配到

黄色Queue：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配

4.5.2. 代码

1）生产者

使用topic类型的Exchange

package com.itheima.producer;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 发送消息
*/
public class Producer_Topics {
  public static void main(String[] args) throws IOException, TimeoutException {
      //1.创建连接工厂
      ConnectionFactory factory = new ConnectionFactory();
      //2. 设置参数
      factory.setHost("172.16.98.133");//ip  默认值 localhost
      factory.setPort(5672); //端口  默认值 5672
      factory.setVirtualHost("/itcast");//虚拟机 默认值/
      factory.setUsername("heima");//用户名 默认 guest
      factory.setPassword("heima");//密码 默认值 guest
      //3. 创建连接 Connection
      Connection connection = factory.newConnection();
      //4. 创建Channel
      Channel channel = connection.createChannel();
     /*
     exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
     参数：
      1. exchange:交换机名称
      2. type:交换机类型
          DIRECT("direct"),：定向
          FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
          TOPIC("topic"),通配符的方式
          HEADERS("headers");参数匹配
      3. durable:是否持久化
      4. autoDelete:自动删除
      5. internal：内部使用。 一般false
      6. arguments：参数
      */
     String exchangeName = "test_topic";
      //5. 创建交换机
      channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);
      //6. 创建队列
      String queue1Name = "test_topic_queue1";
      String queue2Name = "test_topic_queue2";
      channel.queueDeclare(queue1Name,true,false,false,null);
      channel.queueDeclare(queue2Name,true,false,false,null);
      //7. 绑定队列和交换机
      /*
      queueBind(String queue, String exchange, String routingKey)
      参数：
          1. queue：队列名称
          2. exchange：交换机名称
          3. routingKey：路由键，绑定规则
              如果交换机的类型为fanout ，routingKey设置为""
       */
      // routing key  系统的名称.日志的级别。
      //=需求： 所有error级别的日志存入数据库，所有order系统的日志存入数据库
      channel.queueBind(queue1Name,exchangeName,"#.error");
      channel.queueBind(queue1Name,exchangeName,"order.*");
      channel.queueBind(queue2Name,exchangeName,"*.*");
      String body = "日志信息：张三调用了findAll方法...日志级别：info...";
      //8. 发送消息
      channel.basicPublish(exchangeName,"goods.error",null,body.getBytes());
      //9. 释放资源
      channel.close();
      connection.close();
  }
}

2）消费者1

接收两种类型的消息：更新商品和删除商品

package com.itheima.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer_Topic1 {
  public static void main(String[] args) throws IOException, TimeoutException {
      //1.创建连接工厂
      ConnectionFactory factory = new ConnectionFactory();
      //2. 设置参数
      factory.setHost("172.16.98.133");//ip  默认值 localhost
      factory.setPort(5672); //端口  默认值 5672
      factory.setVirtualHost("/itcast");//虚拟机 默认值/
      factory.setUsername("heima");//用户名 默认 guest
      factory.setPassword("heima");//密码 默认值 guest
      //3. 创建连接 Connection
      Connection connection = factory.newConnection();
      //4. 创建Channel
      Channel channel = connection.createChannel();
      String queue1Name = "test_topic_queue1";
      String queue2Name = "test_topic_queue2";

      // 接收消息
      Consumer consumer = new DefaultConsumer(channel){
          /*
              回调方法，当收到消息后，会自动执行该方法
              1. consumerTag：标识
              2. envelope：获取一些信息，交换机，路由key...
              3. properties:配置信息
              4. body：数据
           */
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            /*  System.out.println("consumerTag："+consumerTag);
              System.out.println("Exchange："+envelope.getExchange());
              System.out.println("RoutingKey："+envelope.getRoutingKey());
              System.out.println("properties："+properties);*/
              System.out.println("body："+new String(body));
              System.out.println("将日志信息存入数据库.......");
          }
      };
        /*
      basicConsume(String queue, boolean autoAck, Consumer callback)
      参数：
          1. queue：队列名称
          2. autoAck：是否自动确认
          3. callback：回调对象
       */
      channel.basicConsume(queue1Name,true,consumer);
      //关闭资源？不要
  }
}

3）消费者2

接收所有类型的消息：新增商品，更新商品和删除商品。

package com.itheima.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer_Topic2 {
  public static void main(String[] args) throws IOException, TimeoutException {
      //1.创建连接工厂
      ConnectionFactory factory = new ConnectionFactory();
      //2. 设置参数
      factory.setHost("172.16.98.133");//ip  默认值 localhost
      factory.setPort(5672); //端口  默认值 5672
      factory.setVirtualHost("/itcast");//虚拟机 默认值/
      factory.setUsername("heima");//用户名 默认 guest
      factory.setPassword("heima");//密码 默认值 guest
      //3. 创建连接 Connection
      Connection connection = factory.newConnection();
      //4. 创建Channel
      Channel channel = connection.createChannel();
      String queue1Name = "test_topic_queue1";
      String queue2Name = "test_topic_queue2";

      // 接收消息
      Consumer consumer = new DefaultConsumer(channel){
          /*
              回调方法，当收到消息后，会自动执行该方法
              1. consumerTag：标识
              2. envelope：获取一些信息，交换机，路由key...
              3. properties:配置信息
              4. body：数据
           */
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            /*  System.out.println("consumerTag："+consumerTag);
              System.out.println("Exchange："+envelope.getExchange());
              System.out.println("RoutingKey："+envelope.getRoutingKey());
              System.out.println("properties："+properties);*/
              System.out.println("body："+new String(body));
              System.out.println("将日志信息打印控制台.......");
          }
      };
           /*
      basicConsume(String queue, boolean autoAck, Consumer callback)
      参数：
          1. queue：队列名称
          2. autoAck：是否自动确认
          3. callback：回调对象
       */
      channel.basicConsume(queue2Name,true,consumer);
      //关闭资源？不要
  }
}

4.5.3. 小结

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能；只是Topic在配置routing key 的时候可以使用通配符，显得更加灵活。

1、声明交换机，并且交换机的类型为 topic 2、topics的工作模式topic，设置 routingkey 是带有通配符的字符串，多个字符串之间用 “ . ” 分割，* ：匹配一个单词， #：匹配0或多个单词、特点：包含路由工作模式，比路由模式更灵活

4.6. 模式总结

RabbitMQ工作模式：1、简单模式 HelloWorld一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）
2、工作队列模式 Work Queue一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）
3、发布订阅模式 Publish/subscribe需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列
4、路由模式 Routing需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列
5、通配符模式 Topic需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

5. Spring 整合RabbitMQ

5.1. 搭建生产者工程

5.1.1. 创建工程

5.1.2. 添加依赖

修改pom.xml文件内容为如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.itheima</groupId>
    <artifactId>spring-rabbitmq-producer</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>2.1.8.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>
    </dependencies>
</project>

5.1.3. 配置整合

创建spring-rabbitmq-producer\src\main\resources\properties\rabbitmq.properties连接参数等配置文件；

rabbitmq.host=172.16.98.128
rabbitmq.port=5672
rabbitmq.username=heima
rabbitmq.password=heima
rabbitmq.virtual-host=/itcast

创建 spring-rabbitmq-producer\src\main\resources\spring\spring-rabbitmq.xml 整合配置文件；

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/context
    https://www.springframework.org/schema/context/spring-context.xsd
    http://www.springframework.org/schema/rabbit
    http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
 <!--加载配置文件-->
 <context:property-placeholder location="classpath:properties/rabbitmq.properties"/>
 <!-- 定义rabbitmq connectionFactory -->
 <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                            port="${rabbitmq.port}"
                            username="${rabbitmq.username}"
                            password="${rabbitmq.password}"
                            virtual-host="${rabbitmq.virtual-host}"/>
 <!--定义管理交换机、队列-->
 <rabbit:admin connection-factory="connectionFactory"/>
 <!--定义持久化队列，不存在则自动创建；不绑定到交换机则绑定到默认交换机
 默认交换机类型为direct，名字为：""，路由键为队列的名称
 -->
 <rabbit:queue id="spring_queue" name="spring_queue" auto-declare="true"/>
 <!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~广播；所有队列都能收到消息~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -->
 <!--定义广播交换机中的持久化队列，不存在则自动创建-->
 <rabbit:queue id="spring_fanout_queue_1" name="spring_fanout_queue_1" auto-declare="true"/>
 <!--定义广播交换机中的持久化队列，不存在则自动创建-->
 <rabbit:queue id="spring_fanout_queue_2" name="spring_fanout_queue_2" auto-declare="true"/>
 <!--定义广播类型交换机；并绑定上述两个队列-->
 <rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange" auto-declare="true">
     <rabbit:bindings>
         <rabbit:binding queue="spring_fanout_queue_1"/>
         <rabbit:binding queue="spring_fanout_queue_2"/>
     </rabbit:bindings>
 </rabbit:fanout-exchange>
 <!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~通配符；*匹配一个单词，#匹配多个单词 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -->
 <!--定义广播交换机中的持久化队列，不存在则自动创建-->
 <rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star" auto-declare="true"/>
 <!--定义广播交换机中的持久化队列，不存在则自动创建-->
 <rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/>
 <!--定义广播交换机中的持久化队列，不存在则自动创建-->
 <rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/>
 <rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" auto-declare="true">
     <rabbit:bindings>
         <rabbit:binding pattern="heima.*" queue="spring_topic_queue_star"/>
         <rabbit:binding pattern="heima.#" queue="spring_topic_queue_well"/>
         <rabbit:binding pattern="itcast.#" queue="spring_topic_queue_well2"/>
     </rabbit:bindings>
 </rabbit:topic-exchange>
 <!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
 <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
</beans>

5.1.4. 发送消息

创建测试文件 spring-rabbitmq-producer\src\test\java\com\itheima\rabbitmq\ProducerTest.java

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring/spring-rabbitmq.xml")
public class ProducerTest {
 @Autowired
 private RabbitTemplate rabbitTemplate;
 /**
  * 只发队列消息
  * 默认交换机类型为 direct
  * 交换机的名称为空，路由键为队列的名称
  */
 @Test
 public void queueTest(){
     //路由键与队列同名
     rabbitTemplate.convertAndSend("spring_queue", "只发队列spring_queue的消息。");
 }
 /**
  * 发送广播
  * 交换机类型为 fanout
  * 绑定到该交换机的所有队列都能够收到消息
  */
 @Test
 public void fanoutTest(){
     /**
      * 参数1：交换机名称
      * 参数2：路由键名（广播设置为空）
      * 参数3：发送的消息内容
      */
     rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "发送到spring_fanout_exchange交换机的广播消息");
 }
 /**
  * 通配符
  * 交换机类型为 topic
  * 匹配路由键的通配符，*表示一个单词，#表示多个单词
  * 绑定到该交换机的匹配队列能够收到对应消息
  */
 @Test
 public void topicTest(){
     /**
      * 参数1：交换机名称
      * 参数2：路由键名
      * 参数3：发送的消息内容
      */
     rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.bj", "发送到spring_topic_exchange交换机heima.bj的消息");
     rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.bj.1", "发送到spring_topic_exchange交换机heima.bj.1的消息");
     rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.bj.2", "发送到spring_topic_exchange交换机heima.bj.2的消息");
     rabbitTemplate.convertAndSend("spring_topic_exchange", "itcast.cn", "发送到spring_topic_exchange交换机itcast.cn的消息");
 }
}

5.2. 搭建消费者工程

5.2.1. 创建工程

5.2.2. 添加依赖

修改pom.xml文件内容为如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 <groupId>com.itheima</groupId>
 <artifactId>spring-rabbitmq-consumer</artifactId>
 <version>1.0-SNAPSHOT</version>
 <dependencies>
     <dependency>
         <groupId>org.springframework</groupId>
         <artifactId>spring-context</artifactId>
         <version>5.1.7.RELEASE</version>
     </dependency>
     <dependency>
         <groupId>org.springframework.amqp</groupId>
         <artifactId>spring-rabbit</artifactId>
         <version>2.1.8.RELEASE</version>
     </dependency>
 </dependencies>
</project>

5.2.3. 配置整合

创建spring-rabbitmq-consumer\src\main\resources\properties\rabbitmq.properties连接参数等配置文件；

rabbitmq.host=192.168.12.135
rabbitmq.port=5672
rabbitmq.username=heima
rabbitmq.password=heima
rabbitmq.virtual-host=/itcast

创建 spring-rabbitmq-consumer\src\main\resources\spring\spring-rabbitmq.xml 整合配置文件；

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/context
    https://www.springframework.org/schema/context/spring-context.xsd
    http://www.springframework.org/schema/rabbit
    http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
 <!--加载配置文件-->
 <context:property-placeholder location="classpath:properties/rabbitmq.properties"/>
 <!-- 定义rabbitmq connectionFactory -->
 <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                            port="${rabbitmq.port}"
                            username="${rabbitmq.username}"
                            password="${rabbitmq.password}"
                            virtual-host="${rabbitmq.virtual-host}"/>
 <bean id="springQueueListener" class="com.itheima.rabbitmq.listener.SpringQueueListener"/>
 <bean id="fanoutListener1" class="com.itheima.rabbitmq.listener.FanoutListener1"/>
 <bean id="fanoutListener2" class="com.itheima.rabbitmq.listener.FanoutListener2"/>
 <bean id="topicListenerStar" class="com.itheima.rabbitmq.listener.TopicListenerStar"/>
 <bean id="topicListenerWell" class="com.itheima.rabbitmq.listener.TopicListenerWell"/>
 <bean id="topicListenerWell2" class="com.itheima.rabbitmq.listener.TopicListenerWell2"/>
 <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true">
     <rabbit:listener ref="springQueueListener" queue-names="spring_queue"/>
     <rabbit:listener ref="fanoutListener1" queue-names="spring_fanout_queue_1"/>
     <rabbit:listener ref="fanoutListener2" queue-names="spring_fanout_queue_2"/>
     <rabbit:listener ref="topicListenerStar" queue-names="spring_topic_queue_star"/>
     <rabbit:listener ref="topicListenerWell" queue-names="spring_topic_queue_well"/>
     <rabbit:listener ref="topicListenerWell2" queue-names="spring_topic_queue_well2"/>
 </rabbit:listener-container>
</beans>

5.2.4. 消息监听器

1）队列监听器

创建 spring-rabbitmq-consumer\src\main\java\com\itheima\rabbitmq\listener\SpringQueueListener.java

public class SpringQueueListener implements MessageListener {
 public void onMessage(Message message) {
     try {
         String msg = new String(message.getBody(), "utf-8");
         System.out.printf("接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",
                 message.getMessageProperties().getReceivedExchange(),
                 message.getMessageProperties().getReceivedRoutingKey(),
                 message.getMessageProperties().getConsumerQueue(),
                 msg);
     } catch (Exception e) {
         e.printStackTrace();
     }
 }
}

2）广播监听器1

创建 spring-rabbitmq-consumer\src\main\java\com\itheima\rabbitmq\listener\FanoutListener1.java

public class FanoutListener1 implements MessageListener {
 public void onMessage(Message message) {
     try {
         String msg = new String(message.getBody(), "utf-8");
         System.out.printf("广播监听器1：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",
                 message.getMessageProperties().getReceivedExchange(),
                 message.getMessageProperties().getReceivedRoutingKey(),
                 message.getMessageProperties().getConsumerQueue(),
                 msg);
     } catch (Exception e) {
         e.printStackTrace();
     }
 }
}

3）广播监听器2

创建 spring-rabbitmq-consumer\src\main\java\com\itheima\rabbitmq\listener\FanoutListener2.java

public class FanoutListener2 implements MessageListener {
 public void onMessage(Message message) {
     try {
         String msg = new String(message.getBody(), "utf-8");
         System.out.printf("广播监听器2：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",
                 message.getMessageProperties().getReceivedExchange(),
                 message.getMessageProperties().getReceivedRoutingKey(),
                 message.getMessageProperties().getConsumerQueue(),
                 msg);
     } catch (Exception e) {
         e.printStackTrace();
     }
 }
}

4）星号通配符监听器

创建 spring-rabbitmq-consumer\src\main\java\com\itheima\rabbitmq\listener\TopicListenerStar.java

public class TopicListenerStar implements MessageListener {
 public void onMessage(Message message) {
     try {
         String msg = new String(message.getBody(), "utf-8");
         System.out.printf("通配符*监听器：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",
                 message.getMessageProperties().getReceivedExchange(),
                 message.getMessageProperties().getReceivedRoutingKey(),
                 message.getMessageProperties().getConsumerQueue(),
                 msg);
     } catch (Exception e) {
         e.printStackTrace();
     }
 }
}

5）井号通配符监听器

创建 spring-rabbitmq-consumer\src\main\java\com\itheima\rabbitmq\listener\TopicListenerWell.java

public class TopicListenerWell implements MessageListener {
 public void onMessage(Message message) {
     try {
         String msg = new String(message.getBody(), "utf-8");
         System.out.printf("通配符#监听器：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",
                 message.getMessageProperties().getReceivedExchange(),
                 message.getMessageProperties().getReceivedRoutingKey(),
                 message.getMessageProperties().getConsumerQueue(),
                 msg);
     } catch (Exception e) {
         e.printStackTrace();
     }
 }
}

6）井号通配符监听器2

创建 spring-rabbitmq-consumer\src\main\java\com\itheima\rabbitmq\listener\TopicListenerWell2.java

public class TopicListenerWell2 implements MessageListener {
 public void onMessage(Message message) {
     try {
         String msg = new String(message.getBody(), "utf-8");
         System.out.printf("通配符#监听器2：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",
                 message.getMessageProperties().getReceivedExchange(),
                 message.getMessageProperties().getReceivedRoutingKey(),
                 message.getMessageProperties().getConsumerQueue(),
                 msg);
     } catch (Exception e) {
         e.printStackTrace();
     }
 }
}

6. Spring Boot整合RabbitMQ

6.1. 简介

在spring boot项目中只需要引入对应的amqp启动器依赖即可，方便的使用RabbitTemplate发送消息，使用注解接收消息。
一般在开发过程中：
生产者工程：

application.yml文件配置RabbitMQ相关信息；
在生产者工程中编写配置类，用于创建交换机和队列，并进行绑定
注入RabbitTemplate对象，通过RabbitTemplate对象发送消息到交换机

消费者工程：

application.yml文件配置RabbitMQ相关信息

创建消息处理类，用于接收队列中的消息并进行处理

5.2. 搭建生产者工程

5.2.1. 创建工程

创建生产者工程producer-springboot

5.2.2. 添加依赖

修改pom.xml文件内容为如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 <parent>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-parent</artifactId>
     <version>2.1.4.RELEASE</version>
 </parent>
 <groupId>com.itheima</groupId>
 <artifactId>springboot-rabbitmq-producer</artifactId>
 <version>1.0-SNAPSHOT</version>
 <dependencies>
     <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-amqp</artifactId>
     </dependency>
     <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-test</artifactId>
     </dependency>
 </dependencies>
</project>

5.2.3. 启动类

package com.itheima.rabbitmq;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ProducerApplication {
 public static void main(String[] args) {
     SpringApplication.run(ProducerApplication.class);
 }
}

5.2.4. 配置RabbitMQ

1）配置文件

创建application.yml，内容如下：

spring:
rabbitmq:
 host: localhost
 port: 5672
 virtual-host: /itcast
 username: heima
 password: heima

2）绑定交换机和队列【掌握】

创建RabbitMQ队列与交换机绑定的配置类com.itheima.rabbitmq.config.RabbitMQConfig

package com.itheima.rabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
 public static final String EXCHANGE_NAME = "boot_topic_exchange";
 public static final String QUEUE_NAME = "boot_queue";
 //1.交换机
 @Bean("bootExchange")
 public Exchange bootExchange(){
     return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
 }
 //2.Queue 队列
 @Bean("bootQueue")
 public Queue bootQueue(){
     return QueueBuilder.durable(QUEUE_NAME).build();
 }
 //3. 队列和交互机绑定关系 Binding
 /*
     1. 知道哪个队列
     2. 知道哪个交换机
     3. routing key
  */
 @Bean
 public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") Exchange exchange){
     return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
 }
}

3) 发送消息

package com.itheima.test;
import com.itheima.rabbitmq.config.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@SpringBootTest
@RunWith(SpringRunner.class)
public class ProducerTest {
 //1.注入RabbitTemplate
 @Autowired
 private RabbitTemplate rabbitTemplate;
 @Test
 public void testSend(){
     rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"boot.haha","boot mq hello~~~");
 }
}

5.3. 搭建消费者工程

5.3.1. 创建工程

创建消费者工程consumer-springboot

5.3.2. 添加依赖

修改pom.xml文件内容为如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 <parent>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-parent</artifactId>
     <version>2.1.4.RELEASE</version>
 </parent>
 <groupId>com.itheima</groupId>
 <artifactId>springboot-rabbitmq-consumer</artifactId>
 <version>1.0-SNAPSHOT</version>
 <dependencies>
     <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-amqp</artifactId>
     </dependency>
 </dependencies>
</project>

5.3.3. 启动类

package com.itheima.rabbitmq;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ConsumerApplication {
 public static void main(String[] args) {
     SpringApplication.run(ConsumerApplication.class);
 }
}

5.3.4. 配置RabbitMQ

创建application.yml，内容如下：

spring:
rabbitmq:
 host: localhost
 port: 5672
 virtual-host: /itcast
 username: heima
 password: heima

5.3.5. 消息监听处理类

编写消息监听器com.itheima.rabbitmq.listener.MyListener

package com.itheima.consumerspringboot;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class RabbimtMQListener {
 /**
  * 监听某个队列的消息
  * @param message 接收到的消息
  */
 @RabbitListener(queues = "boot_queue")
 public void ListenerQueue(Message message){
     //System.out.println(message);
     System.out.println(new String(message.getBody()));
 }
}

5.4. 小结

SpringBoot提供了快速整合RabbitMQ的方式
基本信息再yml中配置，队列交互机以及绑定关系在配置类中使用Bean的方式配置
生产端直接注入RabbitTemplate完成消息发送
消费端直接使用@RabbitListener完成消息接收

RabbitMQ 高级特性

消息可靠性投递
在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。
confirm 确认模式
return 退回模式

rabbitmq 整个消息投递的路径为：
producer —-> rabbitmq broker —-> exchange —-> queue —-> consumer

消息从 producer 到 exchange 则会返回一个 confirmCallback 回调函数。
消息从 exchange 到 queue 投递失败则会返回一个 returnCallback 回调函数。

我们将利用这两个 callback 控制消息的可靠性投递

confirm确认模式代码实现

创建maven工程，消息的生产者工程，项目模块名称：rabbitmq-producer-spring
添加依赖
```xml
org.springframework spring-context 5.1.7.RELEASE org.springframework.amqp spring-rabbit 2.1.8.RELEASE junit junit 4.12 org.springframework spring-test 5.1.7.RELEASE

org.apache.maven.plugins maven-compiler-plugin 3.8.0 1.8 1.8


3. 在 resources 目录下创建 rabbitmq.properties 配置文件，添加链接RabbitMQ相关信息
```java
rabbitmq.host=localhost
rabbitmq.port=5672
rabbitmq.username=guest
rabbitmq.password=guest
rabbitmq.virtual-host=/

在 resources 目录下创建 spring-rabbitmq-producer.xml 配置文件，添加以下配置 ```xml <?xml version=”1.0” encoding=”UTF-8”?> <beans xmlns=”http://www.springframework.org/schema/beans“

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans
          http://www.springframework.org/schema/beans/spring-beans.xsd
          http://www.springframework.org/schema/context
          https://www.springframework.org/schema/context/spring-context.xsd
          http://www.springframework.org/schema/rabbit
          http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

<rabbit:connection-factory id=”connectionFactory” host=”${rabbitmq.host}”

                        port="${rabbitmq.port}"
                        username="${rabbitmq.username}"
                        password="${rabbitmq.password}"
                        virtual-host="${rabbitmq.virtual-host}"

                        publisher-confirms="true"
                        />

 <rabbit:bindings>
     <rabbit:binding queue="test_queue_confirm" key="confirm">                           </rabbit:binding>
 </rabbit:bindings>


5. 编写测试代码
```java
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {

  @Autowired
  private RabbitTemplate rabbitTemplate;

  /**
     * 确认模式：
     * 步骤：
     * 1. 确认模式开启：ConnectionFactory中开启publisher-confirms="true"
     * 2. 在rabbitTemplate定义ConfirmCallBack回调函数
     */
  @Test
  public void testConfirm() {

    //2. 定义回调 **
    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
      /**
             *
             * @param correlationData 相关配置信息
             * @param ack   exchange交换机 是否成功收到了消息。true 成功，false代表失败
             * @param cause 失败原因
             */
      @Override
      public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("confirm方法被执行了....");

        if (ack) {
          //接收成功
          System.out.println("接收成功消息" + cause);
        } else {
          //接收失败
          System.out.println("接收失败消息" + cause);
          //做一些处理，让消息再次发送。
        }
      }
    });

    //3. 发送消息
    rabbitTemplate.convertAndSend("test_exchange_confirm111", "confirm", "message confirm....");
  }
}

测试结果

return退回模式代码实现

回退模式：当消息发送给Exchange后，Exchange路由到Queue失败是才会执行ReturnCallBack，具体实现如下：

在 spring-rabbitmq-producer.xml 配置文件，在 rabbit:connection-factory节点添加配置：
```
publisher-returns="true"
```

编写测试方法

/**
* 步骤：
* 1. 开启回退模式:publisher-returns="true"
* 2. 设置ReturnCallBack
* 3. 设置Exchange处理消息的模式：
*  1. 如果消息没有路由到Queue，则丢弃消息（默认）
*  2. 如果消息没有路由到Queue，返回给消息发送方ReturnCallBack
*/
@Test
public void testReturn() {
 //设置交换机处理失败消息的模式
 rabbitTemplate.setMandatory(true);
 //2.设置ReturnCallBack
 rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
     /**
      * 只有转发失败的时候回执行该方法
      * @param message   消息对象
      * @param replyCode 错误码
      * @param replyText 错误信息
      * @param exchange  交换机
      * @param routingKey 路由键
      */
     @Override
     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
         System.out.println("return 执行了....");
         System.out.println(message);
         System.out.println(replyCode);
         System.out.println(replyText);
         System.out.println(exchange);
         System.out.println(routingKey);
         //处理
     }
 });
 //3. 发送消息   
 rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "message confirm....");
}

设置 routingKey 为一个不符合规则的key，观察控制台打印结果。

小结

对于确认模式：

设置ConnectionFactory的publisher-confirms=”true” 开启确认模式。
使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。

对于退回模式

设置ConnectionFactory的publisher-returns=”true” 开启退回模式。
使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，如果设置了rabbitTemplate.setMandatory(true)参数，则会将消息退回给producer。并执行回调函数returnedMessage。

在RabbitMQ中也提供了事务机制，但是性能较差，此处不做讲解。
使用channel列方法，完成事务控制：
txSelect(), 用于将当前channel设置成transaction模式
txCommit()，用于提交事务
txRollback(),用于回滚事务

Consumer ACK

ack指 Acknowledge，确认。表示消费端收到消息后的确认方式。
有三种确认方式：
• 自动确认：acknowledge=”none“
• 手动确认：acknowledge=”manual“
• 根据异常情况确认：acknowledge=”auto“，（这种方式使用麻烦，不作讲解）
其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。
如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

代码实现

创建maven工程，消息的消费者工程，项目模块名称：rabbitmq-consumer-spring
添加依赖 ```xml org.springframework spring-context 5.1.7.RELEASE org.springframework.amqp spring-rabbit 2.1.8.RELEASE junit junit 4.12 org.springframework spring-test 5.1.7.RELEASE

org.apache.maven.plugins maven-compiler-plugin 3.8.0 1.8 1.8


3. 在 resources 目录下创建 rabbitmq.properties 配置文件，添加链接RabbitMQ相关信息
```bash
rabbitmq.host=172.16.98.133
rabbitmq.port=5672
rabbitmq.username=guest
rabbitmq.password=guest
rabbitmq.virtual-host=/

在 resources 目录下创建 spring-rabbitmq-consumer.xml 配置文件，添加以下配置 ```xml <?xml version=”1.0” encoding=”UTF-8”?> <beans xmlns=”http://www.springframework.org/schema/beans“

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans
      http://www.springframework.org/schema/beans/spring-beans.xsd
      http://www.springframework.org/schema/context
      https://www.springframework.org/schema/context/spring-context.xsd
      http://www.springframework.org/schema/rabbit
      http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

<context:component-scan base-package="com.itheima.listener" />

<!--定义监听器容器  添加  acknowledge="manual" 手动-->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" >
    <rabbit:listener ref="ackListener" queue-names="test_queue_confirm">    
    </rabbit:listener>
</rabbit:listener-container>


5. 编写ackListener 监听类实现ChannelAwareMessageListener接口
```java
package com.itheima.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * Consumer ACK机制：
 *  1. 设置手动签收。acknowledge="manual"
 *  2. 让监听器类实现ChannelAwareMessageListener接口
 *  3. 如果消息成功处理，则调用channel的 basicAck()签收
 *  4. 如果消息处理失败，则调用channel的basicNack()拒绝签收，broker重新发送给consumer
 */
@Component
public class AckListener implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            //1.接收转换消息
            System.out.println(new String(message.getBody()));

            //2. 处理业务逻辑
            System.out.println("处理业务逻辑...");
            //int i = 3/0;//出现错误
            //3. 手动签收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();

            //4.拒绝签收
            /*
            第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
             */
            channel.basicNack(deliveryTag,true,true);
            // 了解
            //channel.basicReject(deliveryTag,true);
        }
    }
}

编写测试类，启动容器监听MQ队列

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
public class ConsumerTest {

 @Test
 public void test(){
     while (true){

     }
 }
}

小结

在rabbit:listener-container标签中设置acknowledge属性，设置ack方式 none：自动确认，manual：手动确认
如果在消费端没有出现异常，则调用channel.basicAck(deliveryTag,false);方法确认签收消息
如果出现异常，则在catch中调用 basicNack或 basicReject，拒绝消息，让MQ重新发送消息。

如何保证消息的高可靠性传输？ 1.持久化 • exchange要持久化 • queue要持久化 • message要持久化 2.生产方确认Confirm、return 回退模式 3.消费方确认Ack 4.Broker高可用

消费端限流

如上图所示：如果在A系统中需要维护相关的业务功能，可能需要将A系统的服务停止，那么这个时候消息的生产者还是一直会向MQ中发送待处理的消息，消费者此时服务已经关闭，导致大量的消息都会在MQ中累积。如果当A系统成功启动后，默认情况下消息的消费者会一次性将MQ中累积的大量的消息全部拉取到自己的服务，导致服务在短时间内会处理大量的业务，可能会导致系统服务的崩溃。所以消费端限流是非常有必要的。
可以通过MQ中的 listener-container 配置属性perfetch = 1,表示消费端每次从mq拉去一条消息来消费，直到手动确认消费完毕后，才会继续拉去下一条消息。

代码实现

编写 QosListener 监听类，保证当前的监听类消息处理机制是 ACK 为手动方式

@Component
public class QosListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        Thread.sleep(1000);
        //1.获取消息
        System.out.println(new String(message.getBody()));
        //2. 处理业务逻辑
        //3. 签收
      channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
    }
}

在配置文件的 listener-container 配置属性中添加配置
```
<rabbit:listener-container connection-factory="connectionFactory" 
 acknowledge="manual" prefetch="1" >
```
配置说明： perfetch = 1,表示消费端每次从mq拉去一条消息来消费，直到手动确认消费完毕后，才会继续拉去下一条消息。

小结

在rabbit:listener-container 中配置 prefetch属性设置消费端一次拉取多少消息
消费端的确认模式一定为手动确认。acknowledge=”manual”
TTL

TTL 全称 Time To Live（存活时间/过期时间）。当消息到达存活时间后，还没有被消费，会被自动清除。

RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。代码实现

设置队列的过期时间

在消息的生产方中，在 spring-rabbitmq-producer.xml 配置文件中，添加如下配置： ```java

 <!--x-message-ttl指队列的过期时间-->
 <entry key="x-message-ttl" value="100000" value-type="java.lang.Integer"/>


2. 编写发送消息测试方法
```java
@Test
public void testTtl() {

    for (int i = 0; i < 10; i++) {
        // 发送消息
        rabbitTemplate.convertAndSend("test_exchange_ttl",
                                      "ttl.hehe", "message ttl....");
    }
}

设置单个消息的过期时间(了解)

编写代码测试，并且设置队列的过期时间为100s，单个消息的过期时间为5s

@Test
public void testTtl() {

  // 消息后处理对象，设置一些消息的参数信息
    MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {

        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            //1.设置message的信息
            message.getMessageProperties().setExpiration("5000");//消息的过期时间
            //2.返回该消息
            return message;
        }
    };

    //消息单独过期
    rabbitTemplate.convertAndSend("test_exchange_ttl", 
                                  "ttl.hehe", "message ttl....",messagePostProcessor);

    for (int i = 0; i < 10; i++) {
        if(i == 5){
            //消息单独过期
            rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....",messagePostProcessor);
        }else{
            //不过期的消息
            rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....");

        }

    }
}

死信队列

死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX。

消息成为死信的三种情况：

队列消息长度到达限制；
消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；
原队列存在消息过期设置，消息到达超时时间未被消费；

队列绑定死信交换机：

给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key

代码实现

在消息的生产方中，在 spring-rabbitmq-producer.xml 配置文件中，添加如下配置：

声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)

<rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
</rabbit:queue>
<rabbit:topic-exchange name="test_exchange_dlx">
<rabbit:bindings>
   <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>

声明死信队列(queue_dlx)和死信交换机(exchange_dlx)

<rabbit:queue name="queue_dlx" id="queue_dlx"></rabbit:queue>
<rabbit:topic-exchange name="exchange_dlx">
<rabbit:bindings>
   <rabbit:binding pattern="dlx.#" queue="queue_dlx"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>

正常队列绑定死信交换机，并设置相关参数信息

<rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
<!--3. 正常队列绑定死信交换机-->
<rabbit:queue-arguments>
   <!--3.1 x-dead-letter-exchange：死信交换机名称-->
   <entry key="x-dead-letter-exchange" value="exchange_dlx" />

   <!--3.2 x-dead-letter-routing-key：发送给死信交换机的routingkey-->
   <entry key="x-dead-letter-routing-key" value="dlx.hehe" />

   <!--4.1 设置队列的过期时间 ttl-->
   <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" />
   <!--4.2 设置队列的长度限制 max-length -->
   <entry key="x-max-length" value="10" value-type="java.lang.Integer" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="test_exchange_dlx">
<rabbit:bindings>
   <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>

编写测试方法

    /**
    * 发送测试死信消息：
    *  1. 过期时间
    *  2. 长度限制
    *  3. 消息拒收
    */
    @Test 
    public void testDlx(){
    //1. 测试过期时间，死信消息
    rabbitTemplate.convertAndSend("test_exchange_dlx",
    "test.dlx.haha","我是一条消息，我会死吗？"); 
    //2. 测试长度限制后，消息死信
   for (int i = 0; i < 20; i++) {
        rabbitTemplate.convertAndSend("test_exchange_dlx",
                                      "test.dlx.haha","我是一条消息，我会死吗？");
    }

    //3. 测试消息拒收
    rabbitTemplate.convertAndSend("test_exchange_dlx",
                                  "test.dlx.haha","我是一条消息，我会死吗？");

}

小结

死信交换机和死信队列和普通的没有区别
当消息成为死信后，如果该队列绑定了死信交换机，则消息会被死信交换机重新路由到死信队列
消息成为死信的三种情况：
- 队列消息长度到达限制；
- 消费者拒接消费消息，并且不重回队列；
- 原队列存在消息过期设置，消息到达超时时间未被消费；
  延迟队列

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。

提出需求：

下单后，30分钟未支付，取消订单，回滚库存。
新用户注册成功7天后，发送短信问候。

实现方式：

定时器
延迟队列

注意：在RabbitMQ中并未提供延迟队列功能。
但是可以使用：TTL+死信队列 组合实现延迟队列的效果。

代码实现

在消息的生产方中，在 spring-rabbitmq-producer.xml 配置文件中，添加如下配置： ```xml

 <entry key="x-dead-letter-exchange" value="order_exchange_dlx" />
 <entry key="x-dead-letter-routing-key" value="dlx.order.cancel" />
 <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" />

 <rabbit:binding pattern="order.#" queue="order_queue"></rabbit:binding>


2. 编写测试方法
```java
@Test
public  void testDelay() throws InterruptedException {
    //1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息
    rabbitTemplate.convertAndSend("order_exchange",
                                  "order.msg","订单信息：id=1,time=2019年");

    //2.打印倒计时10秒
    for (int i = 10; i > 0 ; i--) {
        System.out.println(i+"...");
        Thread.sleep(1000);
    }
}

小结

延迟队列指消息进入队列后，可以被延迟一定时间，再进行消费。
RabbitMQ没有提供延迟队列功能，但是可以使用： TTL + DLX 来实现延迟队列效果。
日志与监控
RabbitMQ日志
RabbitMQ默认日志存放路径： /var/log/rabbitmq/rabbit@xxx.log
RabbitMQ 日志所在的目录：
RabbitMQ日志详细信息：

日志包含了RabbitMQ的版本号、Erlang的版本号、RabbitMQ服务节点名称、cookie的hash值、RabbitMQ配置文件地址、内存限制、磁盘限制、默认账户guest的创建以及权限配置等等。

web管控台监控

直接访问当前的IP:15672，输入用户名和密码（默认是 guest），就可以查看RabbitMQ的管理控制台。当然也可通过命令的形式来查看。如下：

查看队列：rabbitmqctl list_queues

对应管理控制台的页面如下：

查看用户： rabbitmqctl list_users
查看连接：rabbitmqctl list_connections

其它相关命令（了解）：
查看exchanges：rabbitmqctl list_exchanges
查看消费者信息：rabbitmqctl list_consumers
查看环境变量：rabbitmqctl environment
查看未被确认的队列：rabbitmqctl list_queues name messages_unacknowledged
查看单个队列的内存使用：rabbitmqctl list_queues name memory
查看准备就绪的队列：rabbitmqctl list_queues name messages_ready

消息追踪

在使用任何消息中间件的过程中，难免会出现某条消息异常丢失的情况。对于RabbitMQ而言，可能是因为生产者或消费者与RabbitMQ断开了连接，而它们与RabbitMQ又采用了不同的确认机制；也有可能是因为交换器与队列之间不同的转发策略；甚至是交换器并没有与任何队列进行绑定，生产者又不感知或者没有采取相应的措施；另外RabbitMQ本身的集群策略也可能导致消息的丢失。这个时候就需要有一个较好的机制跟踪记录消息的投递过程，以此协助开发和运维人员进行问题的定位。

在RabbitMQ中可以使用Firehose和rabbitmq_tracing插件功能来实现消息追踪。

消息追踪-Firehose

firehose的机制是将生产者投递给rabbitmq的消息，rabbitmq投递给消费者的消息按照指定的格式发送到默认的exchange上。这个默认的exchange的名称为 amq.rabbitmq.trace，它是一个topic类型的exchange。发送到这个exchange上的消息的routing key为 publish.exchangename 和 deliver.queuename。其中exchangename和queuename为实际exchange和queue的名称，分别对应生产者投递到exchange的消息，和消费者从queue上获取的消息。
注意：打开 trace 会影响消息写入功能，适当打开后请关闭。

rabbitmqctl trace_on：开启Firehose命令

消息追踪验证：

创建一个队列 test_trace，并将当前的队列绑定到 amq.rabbitmq.trace 交换机上，设置RoutingKey为：#

未开启消息追踪之前，我们发送一个消息

**
当前消息发送成功后，在控制台我们可以看到当前消息的具体信息

设置开启消息追踪，在发送一条消息

完整的消息内容：
我们发现当前消息也正常存在，并且开启消息追踪后，会多出一条消息是 amq.rabbitmq.trace 交换机发给当前队列的消息，消息中的内容是比较完整的。

建议：在开发阶段我们可以开启消息追踪，在实际生产环境建议将其关闭 rabbitmqctl trace_off：关闭Firehose命令

消息追踪-rabbitmq_tracing

rabbitmq_tracing和Firehose在实现上如出一辙，只不过rabbitmq_tracing的方式比Firehose多了一层GUI的包装，更容易使用和管理。
启用插件：rabbitmq-plugins enable rabbitmq_tracing
**
发送消息成功后，我们点击日志文件，要求输入RabbitMQ的登录用户名和密码。
建议：在开发阶段我们可以开启消息追踪插件，在实际生产环境不建议建议开启，除非是非常特殊的业务场景，大家根据实际情况选择开启即可。

RabbitMQ应用问题

消息可靠性保障

提出需求：如何能够保证消息的 100% 发送成功？

首先大家要明确任何一个系统都不能保证消息的 100% 投递成功，我们是可以保证消息以最高最可靠的发送给目标方。

在RabbitMQ中采用 消息补充机制 来保证消息的可靠性
步骤分析：

参与部分：消息生产者、消息消费者、数据库、三个队列（Q1、Q2、Q3）、交换机、回调检查服务、定时检查服务

消息的生产者将业务数据存到数据库中
发送消息给队列Q1
消息的生产者等待一定的时间后，在发送一个延迟消息给队列 Q3
消息的消费方监听 Q1 队列消息，成功接收后
消息的消费方会发送一条确认消息给队列Q2
回调检查服务监听队列Q2 发送的确认消息
回调检查服务接收到确认消息后，将消息写入到消息的数据库表中
回调检查服务同时也会监听队列Q3延迟消息，如果接收到消息会和数据库比对消息的唯一标识
如果发现没有接收到确认消息，那么回调检查服务就会远程调用消息生产者，重新发送消息
重新执行 2-7 步骤，保证消息的可靠性传输
如果发送消息和延迟消息都出现异常，定时检查服务会监控消息库中的消息数据，如果发现不一致的消息然后远程调用消息的生产者重新发送消息。
消息幂等性处理
幂等性指一次和多次请求某一个资源，对于资源本身应该具有同样的结果。也就是说，其任意多次执行对资源本身所产生的影响均与一次执行的影响相同。
在MQ中指，消费多条相同的消息，得到与消费该消息一次相同的结果。
在本教程中使用 乐观锁机制 保证消息的幂等操作
RabbitMQ集群搭建

摘要：实际生产应用中都会采用消息队列的集群方案，如果选择RabbitMQ那么有必要了解下它的集群方案原理

一般来说，如果只是为了学习RabbitMQ或者验证业务工程的正确性那么在本地环境或者测试环境上使用其单实例部署就可以了，但是出于MQ中间件本身的可靠性、并发性、吞吐量和消息堆积能力等问题的考虑，在生产环境上一般都会考虑使用RabbitMQ的集群方案。

集群方案的原理

RabbitMQ这款消息队列中间件产品本身是基于Erlang编写，Erlang语言天生具备分布式特性（通过同步Erlang集群各节点的magic cookie来实现）。因此，RabbitMQ天然支持Clustering。这使得RabbitMQ本身不需要像ActiveMQ、Kafka那样通过ZooKeeper分别来实现HA方案和保存集群的元数据。集群是保证可靠性的一种方式，同时可以通过水平扩展以达到增加消息吞吐量能力的目的。

单机多实例部署

由于某些因素的限制，有时候你不得不在一台机器上去搭建一个rabbitmq集群，这个有点类似zookeeper的单机版。真实生成环境还是要配成多机集群的。有关怎么配置多机集群的可以参考其他的资料，这里主要论述如何在单机中配置多个rabbitmq实例。

主要参考官方文档：https://www.rabbitmq.com/clustering.html

首先确保RabbitMQ运行没有问题

[root@super ~]# rabbitmqctl status
Status of node rabbit@super ...
[{pid,10232},
 {running_applications,
     [{rabbitmq_management,"RabbitMQ Management Console","3.6.5"},
      {rabbitmq_web_dispatch,"RabbitMQ Web Dispatcher","3.6.5"},
      {webmachine,"webmachine","1.10.3"},
      {mochiweb,"MochiMedia Web Server","2.13.1"},
      {rabbitmq_management_agent,"RabbitMQ Management Agent","3.6.5"},
      {rabbit,"RabbitMQ","3.6.5"},
      {os_mon,"CPO  CXC 138 46","2.4"},
      {syntax_tools,"Syntax tools","1.7"},
      {inets,"INETS  CXC 138 49","6.2"},
      {amqp_client,"RabbitMQ AMQP Client","3.6.5"},
      {rabbit_common,[],"3.6.5"},
      {ssl,"Erlang/OTP SSL application","7.3"},
      {public_key,"Public key infrastructure","1.1.1"},
      {asn1,"The Erlang ASN1 compiler version 4.0.2","4.0.2"},
      {ranch,"Socket acceptor pool for TCP protocols.","1.2.1"},
      {mnesia,"MNESIA  CXC 138 12","4.13.3"},
      {compiler,"ERTS  CXC 138 10","6.0.3"},
      {crypto,"CRYPTO","3.6.3"},
      {xmerl,"XML parser","1.3.10"},
      {sasl,"SASL  CXC 138 11","2.7"},
      {stdlib,"ERTS  CXC 138 10","2.8"},
      {kernel,"ERTS  CXC 138 10","4.2"}]},
 {os,{unix,linux}},
 {erlang_version,
     "Erlang/OTP 18 [erts-7.3] [source] [64-bit] [async-threads:64] [hipe] [kernel-poll:true]\n"},
 {memory,
     [{total,56066752},
      {connection_readers,0},
      {connection_writers,0},
      {connection_channels,0},
      {connection_other,2680},
      {queue_procs,268248},
      {queue_slave_procs,0},
      {plugins,1131936},
      {other_proc,18144280},
      {mnesia,125304},
      {mgmt_db,921312},
      {msg_index,69440},
      {other_ets,1413664},
      {binary,755736},
      {code,27824046},
      {atom,1000601},
      {other_system,4409505}]},
 {alarms,[]},
 {listeners,[{clustering,25672,"::"},{amqp,5672,"::"}]},
 {vm_memory_high_watermark,0.4},
 {vm_memory_limit,411294105},
 {disk_free_limit,50000000},
 {disk_free,13270233088},
 {file_descriptors,
     [{total_limit,924},{total_used,6},{sockets_limit,829},{sockets_used,0}]},
 {processes,[{limit,1048576},{used,262}]},
 {run_queue,0},
 {uptime,43651},
 {kernel,{net_ticktime,60}}]

停止rabbitmq服务

[root@super sbin]# service rabbitmq-server stop
Stopping rabbitmq-server: rabbitmq-server.

启动第一个节点：

[root@super sbin]# RABBITMQ_NODE_PORT=5673 RABBITMQ_NODENAME=rabbit1 rabbitmq-server start

              RabbitMQ 3.6.5. Copyright (C) 2007-2016 Pivotal Software, Inc.
  ##  ##      Licensed under the MPL.  See http://www.rabbitmq.com/
  ##  ##
  ##########  Logs: /var/log/rabbitmq/rabbit1.log
  ######  ##        /var/log/rabbitmq/rabbit1-sasl.log
  ##########
              Starting broker...
 completed with 6 plugins.

启动第二个节点：

web管理插件端口占用,所以还要指定其web插件占用的端口号。

[root@super ~]# RABBITMQ_NODE_PORT=5674 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15674}]" RABBITMQ_NODENAME=rabbit2 rabbitmq-server start

              RabbitMQ 3.6.5. Copyright (C) 2007-2016 Pivotal Software, Inc.
  ##  ##      Licensed under the MPL.  See http://www.rabbitmq.com/
  ##  ##
  ##########  Logs: /var/log/rabbitmq/rabbit2.log
  ######  ##        /var/log/rabbitmq/rabbit2-sasl.log
  ##########
              Starting broker...
 completed with 6 plugins.

结束命令：

rabbitmqctl -n rabbit1 stop
rabbitmqctl -n rabbit2 stop

rabbit1操作作为主节点：

[root@super ~]# rabbitmqctl -n rabbit1 stop_app  
Stopping node rabbit1@super ...
[root@super ~]# rabbitmqctl -n rabbit1 reset     
Resetting node rabbit1@super ...
[root@super ~]# rabbitmqctl -n rabbit1 start_app
Starting node rabbit1@super ...
[root@super ~]#

rabbit2操作为从节点：

[root@super ~]# rabbitmqctl -n rabbit2 stop_app
Stopping node rabbit2@super ...
[root@super ~]# rabbitmqctl -n rabbit2 reset
Resetting node rabbit2@super ...
[root@super ~]# rabbitmqctl -n rabbit2 join_cluster rabbit1@'super' ###''内是主机名换成自己的
Clustering node rabbit2@super with rabbit1@super ...
[root@super ~]# rabbitmqctl -n rabbit2 start_app
Starting node rabbit2@super ...

查看集群状态：

[root@super ~]# rabbitmqctl cluster_status -n rabbit1
Cluster status of node rabbit1@super ...
[{nodes,[{disc,[rabbit1@super,rabbit2@super]}]},
 {running_nodes,[rabbit2@super,rabbit1@super]},
 {cluster_name,<<"rabbit1@super">>},
 {partitions,[]},
 {alarms,[{rabbit2@super,[]},{rabbit1@super,[]}]}]

web监控：

集群管理

rabbitmqctl join_cluster {cluster_node} [–ram]
将节点加入指定集群中。在这个命令执行前需要停止RabbitMQ应用并重置节点。

rabbitmqctl cluster_status
显示集群的状态。

rabbitmqctl change_cluster_node_type {disc|ram}
修改集群节点的类型。在这个命令执行前需要停止RabbitMQ应用。

rabbitmqctl forget_cluster_node [–offline]
将节点从集群中删除，允许离线执行。

rabbitmqctl update_cluster_nodes {clusternode}

在集群中的节点应用启动前咨询clusternode节点的最新信息，并更新相应的集群信息。这个和join_cluster不同，它不加入集群。考虑这样一种情况，节点A和节点B都在集群中，当节点A离线了，节点C又和节点B组成了一个集群，然后节点B又离开了集群，当A醒来的时候，它会尝试联系节点B，但是这样会失败，因为节点B已经不在集群中了。

rabbitmqctl cancel_sync_queue [-p vhost] {queue}
取消队列queue同步镜像的操作。

rabbitmqctl set_cluster_name {name}
设置集群名称。集群名称在客户端连接时会通报给客户端。Federation和Shovel插件也会有用到集群名称的地方。集群名称默认是集群中第一个节点的名称，通过这个命令可以重新设置。

RabbitMQ镜像集群配置

上面已经完成RabbitMQ默认集群模式，但并不保证队列的高可用性，尽管交换机、绑定这些可以复制到集群里的任何一个节点，但是队列内容不会复制。虽然该模式解决一项目组节点压力，但队列节点宕机直接导致该队列无法应用，只能等待重启，所以要想在队列节点宕机或故障也能正常应用，就要复制队列内容到集群里的每个节点，必须要创建镜像队列。镜像队列是基于普通的集群模式的，然后再添加一些策略，所以你还是得先配置普通集群，然后才能设置镜像队列，我们就以上面的集群接着做。

设置的镜像队列可以通过开启的网页的管理端Admin->Policies，也可以通过命令。

rabbitmqctl set_policy my_ha “^” ‘{“ha-mode”:”all”}’

Name:策略名称
Pattern：匹配的规则，如果是匹配所有的队列，是^.
Definition:使用ha-mode模式中的all，也就是同步所有匹配的队列。问号链接帮助文档。

负载均衡-HAProxy

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

安装HAProxy

//下载依赖包
yum install gcc vim wget
//上传haproxy源码包
//解压
tar -zxvf haproxy-1.6.5.tar.gz -C /usr/local
//进入目录、进行编译、安装
cd /usr/local/haproxy-1.6.5
make TARGET=linux31 PREFIX=/usr/local/haproxy
make install PREFIX=/usr/local/haproxy
mkdir /etc/haproxy
//赋权
groupadd -r -g 149 haproxy
useradd -g haproxy -r -s /sbin/nologin -u 149 haproxy
//创建haproxy配置文件
mkdir /etc/haproxy
vim /etc/haproxy/haproxy.cfg

3.5.2 配置HAProxy

配置文件路径：/etc/haproxy/haproxy.cfg

#logging options
global
    log 127.0.0.1 local0 info
    maxconn 5120
    chroot /usr/local/haproxy
    uid 99
    gid 99
    daemon
    quiet
    nbproc 20
    pidfile /var/run/haproxy.pid
defaults
    log global

    mode tcp
    option tcplog
    option dontlognull
    retries 3
    option redispatch
    maxconn 2000
    contimeout 5s

     clitimeout 60s
     srvtimeout 15s 
#front-end IP for consumers and producters
listen rabbitmq_cluster
    bind 0.0.0.0:5672

    mode tcp
    #balance url_param userid
    #balance url_param session_id check_post 64
    #balance hdr(User-Agent)
    #balance hdr(host)
    #balance hdr(Host) use_domain_only
    #balance rdp-cookie
    #balance leastconn
    #balance source //ip

    balance roundrobin

        server node1 127.0.0.1:5673 check inter 5000 rise 2 fall 2
        server node2 127.0.0.1:5674 check inter 5000 rise 2 fall 2
listen stats
    bind 172.16.98.133:8100
    mode http
    option httplog
    stats enable
    stats uri /rabbitmq-stats
    stats refresh 5s

启动HAproxy负载

/usr/local/haproxy/sbin/haproxy -f /etc/haproxy/haproxy.cfg
//查看haproxy进程状态
ps -ef | grep haproxy
访问如下地址对mq节点进行监控
http://172.16.98.133:8100/rabbitmq-stats

代码中访问mq集群地址，则变为访问haproxy地址:5672

RabbitMQ

学习目标

1. 消息中间件概述

1.1 什么是消息中间件

1.2 MQ 的优势和劣势

1.2.1 应用解耦

1.2.2 异步提速

1.2.3 削峰填谷

1.2.4 小结

1.3 MQ 的劣势

1.4 消息队列产品

1.5 RabbitMQ

1.7 AMQP 和 JMS

1.7.1. AMQP

1.7.2. JMS

2. 安装及配置RabbitMQ

3. RabbitMQ入门

3.1. 搭建示例工程

3.1.1. 创建工程

3.1.2. 添加依赖

3.2. 编写生产者

3.3. 编写消费者

3.4. 小结

4. RabbitMQ工作模式

4.1. Work queues工作队列模式

4.1.1. 模式说明

4.1.2. 代码

1）生产者

2）消费者1

3）消费者2

4.1.3. 小结

4.2. 订阅模式类型

4.3. Publish/Subscribe发布与订阅模式

4.3.1. 模式说明

4.3.2. 代码

1）生产者

2）消费者1

3）消费者2

4.3.3. 小结

4.4. Routing路由模式

4.4.1. 模式说明

4.4.2. 代码

1）生产者

2）消费者1

3）消费者2

4.4.3. 小结

4.5. Topics通配符模式

4.5.1. 模式说明

4.5.2. 代码

1）生产者

2）消费者1

3）消费者2

4.5.3. 小结

4.6. 模式总结

5. Spring 整合RabbitMQ

5.1. 搭建生产者工程

5.1.1. 创建工程

5.1.2. 添加依赖

5.1.3. 配置整合

5.1.4. 发送消息

5.2. 搭建消费者工程

5.2.1. 创建工程

5.2.2. 添加依赖

5.2.3. 配置整合

5.2.4. 消息监听器

1）队列监听器

2）广播监听器1

3）广播监听器2

4）星号通配符监听器

5）井号通配符监听器

6）井号通配符监听器2

6. Spring Boot整合RabbitMQ

6.1. 简介

5.2. 搭建生产者工程

5.2.1. 创建工程

5.2.2. 添加依赖

5.2.3. 启动类

5.2.4. 配置RabbitMQ

1）配置文件

2）绑定交换机和队列【掌握】