rabbitmq，从入门基础到整合springboot，超详细笔记

前言

什么是RabbitMQ？

MQ(Message Quene) : 翻译为 消息队列,通过典型的 生产者和消费者模型,生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入,轻松的实现系统间解耦。别名为 消息中间件 通过利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。

常见的MQ

老牌的ActiveMQ、RabbitMQ，当下流行的Kafka，阿里巴巴自主开发RocketMQ等。

不同MQ特点

# 1.ActiveMQ
ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。它是一个完全支持JMS规范的的消息中间件。丰富的API,多种集群架构模式让ActiveMQ在业界成为老牌的消息中间件,在中小型企业颇受欢迎!

# 2.Kafka
Kafka是LinkedIn开源的分布式发布-订阅消息系统，目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输。0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务。

# 3.RocketMQ
RocketMQ是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起
源于Kafka，但并不是Kafka的一个Copy，它对消息的可靠传输及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。

# 4.RabbitMQ
RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由（包括点对点和
发布/订阅）、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。

为什么要学习rabbitmq呢？

RabbitMQ比Kafka可靠，Kafka更适合IO高吞吐的处理，一般应用在大数据日志处理或对实时性（少量延迟），可靠性（少量丢数据）要求稍低的场景使用，比如ELK日志收集。

RabbitMQ

简介

基于AMQP协议，erlang语言开发，是部署最广泛的开源消息中间件,是最受欢迎的开源消息中间件之一。

AMQP协议：

AMQP（advanced message queuing protocol）`在2003年时被提出，最早用于解决金融领不同平台之间的消息传递交互问题。顾名思义，AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。以下是AMQP协议模型:

下载安装

因为我这里是部署到远程服务器中，所以这里只讲linux安装运行哈

下载直接从官网下载即可，这里就不详细说明了，如果想要资源的话也可以直接私信我

这里下载了三个包：

# 因为rabbitmq是使用erlang语言写的，所以需要安装erlang
erlang-22.0.7-1.el7.x86_64.rpm
# 安装rabbitmq需要的依赖
socat-1.7.3.2-2.el7.x86_64。rpm
# rabbitmq服务
rabbitmq-server-3.7.18-1.el7.noarch.rpm

然后使用命令行安装

rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm
rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm
rpm -ivh rabbitmq-server-3.7.18-1.el7.noarch.rpm 

都安装完成后，会在以下目录生成配置文件模板

[root@ryan rabbitmq]# find / -name rabbitmq.config.example
/usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example

我们需要将次配置文件模板复制到/etc/rabbitmq/目录下

[root@ryan rabbitmq]# cp /usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example /etc/rabbitmq
[root@ryan rabbitmq]# cd /etc/rabbitmq
[root@ryan rabbitmq]# ls
rabbitmq.config

进入编辑模式，开发来宾模式

[root@ryan rabbitmq]# vim rabbitmq.config
# 删除下图中这一行的注释（百分号）和最后的逗号即可
# 来宾账户密码都是guest

启动rabbitmq中的插件管理

rabbitmq-plugins enable rabbitmq_management

最后启动rabbitmq服务

[root@ryan rabbitmq]# systemctl start rabbitmq-server
# 如果是重启则是restart

可通过命令行查看rabbitmq的服务状态

[root@ryan rabbitmq]# systemctl status rabbitmq-server

如果看到Active：active则说明启动成功，如果Active：dead则说明没有启动成功

如果还没关闭服务器防火墙的一定要先关闭防火墙哦

# 关闭防火墙服务
systemctl disable firewalld
systemctl stop firewalld

最后就可以访问rabbitmq的web后台管理界面了

# 访问地址,rabbitmq的默认端口号是15672
远程服务器主机名:15672

相关配置

命令行：

# 1.服务启动相关
systemctl start|restart|stop|status rabbitmq-server

# 2.管理命令行  用来在不使用web管理界面情况下命令操作RabbitMQ
rabbitmqctl  help  # 可以查看更多命令

# 3.插件管理命令行
rabbitmq-plugins enable|list|disable 

web界面简单介绍

• verview概览

• connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况

• channels：通道，建立连接后，会形成通道，消息的投递获取依赖通道。

• Exchanges：交换机，用来实现消息的路由

• Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列。

实战

从rabbitmq官网可以看到由多种模型

helloworld

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

• P：生产者，也就是要发送消息的程序
• C：消费者：消息的接受者，会一直等待消息到来。
• queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

1、创建1个mave项目，导入相关依赖

<dependencies>
    <!-- amqp-client -->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.9.0</version>
    </dependency>

    <!-- junit -->
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>

    <!--slf4j-->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.25</version>
    </dependency>

</dependencies>

2、编写工具类

/获取连接工具类
public class RabbitMQUtil {

    private static ConnectionFactory connectionFactory;

    static {
        connectionFactory = new ConnectionFactory();
        //设置连接rabbitmq主机
        connectionFactory.setHost("203.195.160.231");
        //设置端口号（不是http端口号）
        connectionFactory.setPort(5672);
        //设置连接哪个虚拟主机
        connectionFactory.setVirtualHost("/ems");
        //设置访问虚拟主机的用户名和密码
        connectionFactory.setUsername("ems");
        connectionFactory.setPassword("123");
    }

    //获取连接对象
    public static Connection getConnection(){
        try {
            return connectionFactory.newConnection();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        return null;
    }

    //关闭连接
    public static void closeChannelAndConnection(Channel channel, Connection connection){
        try{
            if(channel != null) channel.close();
            if(connection != null) connection.close();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

3、编写生产者类

//生产者
public class Provider {

    @Test
    public void testSendMessage() throws IOException, TimeoutException {

        //获取连接对象
        Connection connection = RabbitMQUtil.getConnection();
        //获取连接通道
        Channel channel = connection.createChannel();
        //通道绑定对应的消息队列
        /**
         * 参数1：queue队列名称，如果不存在会自动创建
         * 参数2：durable用来定义队列是否要持久化，如果为false不持久化，在关闭rabbitmq服务器之后，队列会丢失
         *        如果为true持久化，那么rabbitmq重启后队列会恢复，当然消息不一定回复（需要在发布消息时设置）
         * 参数3：exclusive是否独占队列,如果为true，表明只有当前通道能使用该队列，当然我们一般都是为false
         * 参数4：autoDelete是否在消费完成后自动删除队列（需要消费线程断开之后才会被删除）
         * 参数5：arguments额外附加参数
         */
        channel.queueDeclare("hello", true, false, false, null);

        //发布消息
        /**
         * 参数1：交换机名称，此例子不经过交换机
         * 参数2：队列名称
         * 参数3：传递消息额外设置，可设置消息是否持久化，null表示不持久化，MessageProperties.PERSISTENT_TEXT_PLAIN表示持久化消息
         * 参数4：消息的具体内容
         */
        channel.basicPublish("", "hello", MessageProperties.PERSISTENT_TEXT_PLAIN, "hello,rabbitmq".getBytes());

        //关闭通道和连接
        RabbitMQUtil.closeChannelAndConnection(channel, connection);
    }
}

4、运行

然后可以在rabbitmq管理界面看到queue刷新到了消息，再运行一次则会多一条消息

5、编写消费者

//消费者
public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {

        //获取连接对象
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("hello", true, false, false, null);
        //消费消息
        channel.basicConsume("hello", true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body=>" + new String(body));
            }
        });

        //消费者我们一般会让他去一直监听着消息队列，所以一般不会去关闭他
        //运行之后，发现队列中的消息全都消费了，每消费一条消息，都会触发handleDelivery方法
    }
}

控制台信息：

work queue

Work queues，工作队列，也被称为（Task queues），任务模型。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

角色：

• P：生产者：任务的发布者
• C1：消费者-1，领取任务并且完成任务
• C2：消费者-2：领取任务并完成任务

开发生产者

public class Porvider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work queue", true, false, false, null);
        for (int i = 1; i < 10; i++) {
            channel.basicPublish("", "work queue", MessageProperties.PERSISTENT_TEXT_PLAIN, (i + " hello,work queue").getBytes());
        }
        RabbitMQUtil.closeChannelAndConnection(channel, connection);
    }
}

消费者1

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work queue", true, false, false, null);
        channel.basicConsume("work queue", true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1=>" + new String(body) );
            }
        });
    }
}

消费者2

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work queue", true, false, false, null);
        channel.basicConsume("work queue", true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2=>" + new String(body) );
            }
        });
    }
}

先将两个消费者运行起来，最后运行生产者，发现两个消费者是平均小费的

小结：默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。

思考1：循环模式，是从队列中一下子分配消息给消费者的，比如每个消费者5个，消费者1在消费第3个消息的时候，宕机了怎么办？这时候剩下的消息就会丢失，那怎么防止这种现象的出现呢？

思考2：如果一个消费者效率比较快，另一个效率稍微慢一点，怎么实现消费者多劳多得呢？

其实这两个问题都跟一个消息确认机制有关，目前我们都是开着自动确认消息的：也就是下面的参数2

channel.basicConsume("work queue", true, new DefaultConsumer(channel)

如果我们不想一下子直接平均分配给每个消费者的话，我们应该关闭消息自动确认机制，并且管道需要告知队列每次只消费1个消息

同时为了体现其中一个消费者效率比较慢，我们让啊消费者1睡2秒

最后手动确认消息

channel.basicQos(1);//每次只消费1条消息
 channel.basicConsume("work queue", false, new DefaultConsumer(channel){//参数2：false，关闭自动确认消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者1=>" + new String(body) );
                channel.basicAck(envelope.getDeliveryTag(), false);//手动确认消息
            }
        });

测试：发现消费者1只消费了1条消息，其他的由消费者消费

fanout

也称广播

在广播模式下，消息发送流程是这样的：

• 可以有多个消费者
• 每个消费者有自己的queue（队列）
• 每个队列都要绑定到Exchange（交换机）
• 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
• 交换机把消息发送给绑定过的所有队列
• 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

生产者

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //通道绑定交换机
        /**
         * 参数：交换机名称，不存在的话会自动创建
         * 参数2：交换机类型，注意名称是固定的
         */
        channel.exchangeDeclare("logs", "fanout");
        //生产者只需要绑定交换机即可，其他的不需要管的
        //生产消息
        /**
         * 参数1：交换机名称
         * 参数2：路由，在此类型基本没什么用，所以可以空着
         * 参数3：是否持久化消息
         * 参数4：消息体
         */
        channel.basicPublish("logs", "", null, "hello,fanout".getBytes());
        //释放资源
        RabbitMQUtil.closeChannelAndConnection(channel, connection);

    }
}

消费者1/2

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //通道绑定交换机
        channel.exchangeDeclare("logs", "fanout");
        //获取临时队列
        String queueName = channel.queueDeclare().getQueue();
        //绑定交换机和队列
        /**
         * 参数1：临时队列名称
         * 参数2：交换机名称
         * 参数3：路由，在此没什么用，空着
         */
        channel.queueBind(queueName, "logs", "");
        //消费消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1=>" + new String(body));
            }
        });

    }
}

测试：生产者生产消息后，每个消费者都收到了同一条消息

Routing-direct

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下：

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

流程：

图解：

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

生产者：

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        String exchange = "logs_direct";
        /**
         * 参数1：交换机名称
         * 参数2：交换机类型，固定的
         */
        channel.exchangeDeclare(exchange, "direct");
        //声明路由key
        String routeKey = "error";
        //发布消息
        /**
         * 参数1：交换机名称
         * 参数2：路由key
         */
        channel.basicPublish(exchange, routeKey, null, ("direct类型，路由key为" + routeKey +"发来的消息").getBytes());
        //释放资源
        RabbitMQUtil.closeChannelAndConnection(channel, connection);
    }
}

消费者1

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //绑定路由
        String exchange = "logs_direct";
        channel.exchangeDeclare(exchange, "direct");
        //临时队列
        String queue = channel.queueDeclare().getQueue();
        //绑定队列和交换机
        channel.queueBind(queue, exchange, "error");
        channel.queueBind(queue, exchange, "info");
        //消费队列
        channel.basicConsume(queue, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1=>" + new String(body));
            }
        });
    }
}

消费者2：只绑定了info

测试：生产者绑定路由“error”发消息后，只有消费者1消费了

Routing-topic

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

# 统配符
* (star) can substitute for exactly one word.匹配不多不少恰好1个词
# (hash) can substitute for zero or more words.匹配一个或多个词
# 如:
audit.#  匹配audit.irs.corporate或者 audit.irs 等
audit.*  只能匹配 audit.irs

生产者

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("topics", "topic");
        String roteKey = "user.save";
        channel.basicPublish("topics", roteKey, null, "topics发布的消息来了".getBytes());
        RabbitMQUtil.closeChannelAndConnection(channel, connection);
    }
}

消费者1

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("topics", "topic");
        String queue = channel.queueDeclare().getQueue();
        channel.queueBind(queue, "topics", "*.user");
        channel.basicConsume(queue, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1=>" + new String(body));
            }
        });
    }
}

消费者2

channel.queueBind(queue, "topics", "user.#");

测试：因为只有消费者2的通配符符合条件，所以最终只有2消费了消息

Springboot整合RabbitMQ

helloworld

1、新建springboot项目，导入相关依赖

<?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 https://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.3.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.ryan</groupId>
    <artifactId>springboot-rabbitmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

2、配置yml文件

spring:
  application:
    name: springboot-rabbitmq

  rabbitmq:
    host: 203.195.160.231
    username: ems
    password: 123
    virtual-host: /ems
    port: 5672

3、编写生产者

@Test
void testHello(){
    //注意，如果没有对应的消费者，即使执行了生产者，是不会新建队列的
    rabbitTemplate.convertAndSend("hello", "hello发来的信息");
}

4、消费者

@Component
public class Consumer {

    //监听队列，声明队列，可以是单个值，也可以是数组
    @RabbitListener(queuesToDeclare = @Queue("hello"))
    public void receive(String message){
        System.out.println("message=>" + message);
    }

}

测试：message=>hello发来的信息

springboot就是舒服

work queue

1、生产者

@Test
void testWork(){
    for (int i = 0; i < 10; i++) {
        rabbitTemplate.convertAndSend("work","work queue发来的消息" + i);
    }
}

2、消费者

@Component
public class WorkConsumer {

    @RabbitListener(queuesToDeclare = @Queue(value = "work"))
    public void receive1(String message){
        System.out.println("message1=>" + message);
    }

    @RabbitListener(queuesToDeclare = @Queue(value = "work"))
    public void receive2(String message){
        System.out.println("message2=>" + message);
    }

}

测试：略

fanout

1、生产者

@Test
void testFanout(){
    rabbitTemplate.convertAndSend("logs", "", "fanout发来的消息");
}

2、消费者

@Component
public class FanoutConsumer {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,//空表示临时队列
            exchange = @Exchange(value = "logs", type = "fanout")//交换机名称及类型
    ))
    public void receive1(String message){
        System.out.println("message1=>" + message);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,//空表示临时队列
            exchange = @Exchange(value = "logs", type = "fanout")//交换机名称及类型
    ))
    public void receive2(String message){
        System.out.println("message2=>" + message);
    }

}

测试：略

Routing-direct

1、生产者

@Test
void testDirect(){
    rabbitTemplate.convertAndSend("route_direct", "info", "direct交换器传来的消息");
}

2、消费者

@Component
public class DirectConsumer {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,
            exchange = @Exchange(value = "route_direct", type = "direct"),
            key = {"info", "error"}
    ))
    public void receive1(String message){
        System.out.println("message1=>" + message);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,
            exchange = @Exchange(value = "route_direct", type = "direct"),
            key = {"error"}
    ))
    public void receive2(String message){
        System.out.println("message2=>" + message);
    }
}

测试：略

Routing-topics

1、生产者

@Test
void testTopics(){
    rabbitTemplate.convertAndSend("route_topics", "user.save", "topics交换器发来的消息");
}

2、消费者

@Component
public class TopicsConsumer {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,
            key = {"*.save"},
            exchange = @Exchange(value = "route_topics", type = "topic")
    ))
    public void receive1(String message){
        System.out.println("message1=>" + message);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,
            key = {"user.#"},
            exchange = @Exchange(value = "route_topics", type = "topic")
    ))
    public void receive2(String message){
        System.out.println("message2=>" + message);
    }

}

MQ的应用场景

1、异步处理

场景说明：用户注册后，需要发注册邮件和注册短信,传统的做法有两种 1.串行的方式 ：

将注册信息写入数据库后,发送注册邮件,再发送注册短信,以上三个任务全部完成后才返回给客户端。 这有一个问题是,邮件,短信并不是必须的,它只是一个通知,而这种做法让客户端等待没有必要等待的东西.

2.并行的方式：

将注册信息写入数据库后,发送邮件的同时,发送短信,以上三个任务完成后,返回给客户端,并行的方式能提高处理的时间。

• 消息队列:假设三个业务节点分别使用50ms,串行方式使用时间150ms,并行使用时间100ms。虽然并行已经提高的处理时间,但是,前面说过,邮件和短信对我正常的使用网站没有任何影响，客户端没有必要等着其发送完成才显示注册成功,应该是写入数据库后就返回.
• 引入消息队列后，把发送邮件,短信不是必须的业务逻辑异步处理
  
  由此可以看出,引入消息队列后，用户的响应时间就等于写入数据库的时间+写入消息队列的时间(可以忽略不计),引入消息队列后处理后,响应时间是串行的3倍,是并行的2倍

2、应用解耦

场景：双11是购物狂节,用户下单后,订单系统需要通知库存系统,传统的做法就是订单系统调用库存系统的接口.

这种做法有一个缺点:

当库存系统出现故障时,订单就会失败。 订单系统和库存系统高耦合.

引入消息队列 ：

• 订单系统:用户下单后,订单系统完成持久化处理,将消息写入消息队列,返回用户订单下单成功。

• 库存系统:订阅下单的消息,获取下单消息,进行库操作。 就算库存系统出现故障,消息队列也能保证消息的可靠投递,不会导致消息丢失.

3、流量削峰

场景：秒杀活动，一般会因为流量过大，导致应用挂掉,为了解决这个问题，一般在应用前端加入消息队列。

作用：

• 可以控制活动人数，超过此一定阀值的订单直接丢弃(我为什么秒杀一次都没有成功过呢^^)
• 可以缓解短时间的高流量压垮应用(应用程序按自己的最大处理能力获取订单)
  
  1.用户的请求,服务器收到之后,首先写入消息队列,加入消息队列长度超过最大值,则直接抛弃用户请求或跳转到错误页面.

2.秒杀业务根据消息队列中的请求信息，再做后续处理.

RabbitMQ的集群

集群架构

普通集群（副本集群）

默认情况下:RabbitMQ代理操作所需的所有数据/状态都将跨所有节点复制。这方面的一个例外是消息队列，默认情况下，消息队列位于一个节点上，尽管它们可以从所有节点看到和访问。

所以当主节点宕机之后，其他节点队列中的信息也会随之丢失

核心解决问题: 当集群中某一时刻master节点宕机,可以对Quene中信息,进行备份，说白了就是位了解决高可用的问题，需要用到下面的镜像集群

集群搭建步骤：略

镜像集群

镜像队列机制就是将队列在三个节点之间设置主从关系，消息会在三个节点之间进行自动同步，且如果其中一个节点不可用，并不会导致消息丢失或服务不可用的情况，提升MQ集群的整体高可用性。

配置集群架构

# 0.策略说明
	rabbitmqctl set_policy [-p <vhost>] [--priority <priority>] [--apply-to <apply-to>] <name> <pattern>  <definition>
	-p Vhost： 可选参数，针对指定vhost下的queue进行设置
	Name:     policy的名称
	Pattern: queue的匹配模式(正则表达式)
	Definition：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-mode
           		ha-mode:指明镜像队列的模式，有效值为 all/exactly/nodes
                        all：表示在集群中所有的节点上进行镜像
                        exactly：表示在指定个数的节点上进行镜像，节点的个数由ha-params指定
                        nodes：表示在指定的节点上进行镜像，节点名称通过ha-params指定
            	 ha-params：ha-mode模式需要用到的参数
                ha-sync-mode：进行队列中消息的同步方式，有效值为automatic和manual
                priority：可选参数，policy的优先级
                
                 
# 1.查看当前策略
	rabbitmqctl list_policies

# 2.添加策略
	rabbitmqctl set_policy ha-all '^hello' '{"ha-mode":"all","ha-sync-mode":"automatic"}' 
	说明:策略正则表达式为 “^” 表示所有匹配所有队列名称  ^hello:匹配hello开头队列

# 3.删除策略
	rabbitmqctl clear_policy ha-all

# 4.测试集群

---

具体的入门实操就到这里，后面如果有结合实战的我会继续更新，如果觉得有帮助到你的话可以点个赞支持一下哦！