RabbitMQ入门详解

前言

本篇文章将详细介绍rabbitmq的基本概念知识，以及rabbitmq各个工作模式在springboot中如何使用。

文章目录

介绍

简介

RabbitMQ 核心

生产者与消费者

Exchange

Queue

工作模式

简单模式

工作队列模式

发布订阅模式

路由模式

主题模式

SpringBoot中使用RabbitMQ

简单模式

基础配置

生产者

消费者

工作队列模式

发布/订阅模式（广播模式）

路由模式

主题模式

介绍

简介

RabbitMQ 是采用 Erlang 语言实现 AMQP(Advanced Message Queuing Protocol，高级消息队列协议）的消息中间件，它最初起源于金融系统，用于在分布式系统中存储转发消息。

RabbitMQ 发展到今天，被越来越多的人认可，这和它在易用性、扩展性、可靠性和高可用性等方面的卓著表现是分不开的。RabbitMQ 的具体特点可以概括为以下几点：

可靠性： RabbitMQ使用一些机制来保证消息的可靠性，如持久化、传输确认及发布确认等。
灵活的路由： 在消息进入队列之前，通过交换器来路由消息。对于典型的路由功能，RabbitMQ 己经提供了一些内置的交换器来实现。针对更复杂的路由功能，可以将多个交换器绑定在一起，也可以通过插件机制来实现自己的交换器。这个后面会在我们将 RabbitMQ 核心概念的时候详细介绍到。
扩展性： 多个RabbitMQ节点可以组成一个集群，也可以根据实际业务情况动态地扩展集群中节点。
高可用性： 队列可以在集群中的机器上设置镜像，使得在部分节点出现问题的情况下队列仍然可用。
支持多种协议： RabbitMQ 除了原生支持 AMQP 协议，还支持 STOMP、MQTT 等多种消息中间件协议。
多语言客户端： RabbitMQ几乎支持所有常用语言，比如 Java、Python、Ruby、PHP、C#、JavaScript等。
易用的管理界面： RabbitMQ提供了一个易用的用户界面，使得用户可以监控和管理消息、集群中的节点等。在安装 RabbitMQ 的时候会介绍到，安装好 RabbitMQ 就自带管理界面。
插件机制： RabbitMQ 提供了许多插件，以实现从多方面进行扩展，当然也可以编写自己的插件。感觉这个有点类似 Dubbo 的 SPI机制。

RabbitMQ 核心

生产者与消费者

RabbitMQ 整体上是一个生产者与消费者模型，主要负责接收、存储和转发消息。可以把消息传递的过程想象成：当你将一个包裹送到邮局，邮局会暂存并最终将邮件通过邮递员送到收件人的手上，RabbitMQ就好比由邮局、邮箱和邮递员组成的一个系统。

Producer(生产者) :生产消息的一方（邮件投递者）
Consumer(消费者) :消费消息的一方（邮件收件人）

消息一般由 2 部分组成：消息头（或者说是标签 Label）和 消息体。消息体也可以称为 payLoad ,消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括 routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。生产者把消息交由 RabbitMQ 后，RabbitMQ 会根据消息头把消息发送给感兴趣的 Consumer(消费者)。

Exchange

在 RabbitMQ 中，消息并不是直接被投递到Queue中的，中间还必须经过 Exchange(交换机) 这一层，Exchange 会把我们的消息分配到对应的Queue中。

Exchange用来接收生产者发送的消息并将这些消息路由给服务器中的队列中，如果路由不到，或许会返回给 Producer(生产者) ，或许会被直接丢弃掉。这里可以将RabbitMQ中的交换器看作一个简单的实体。

RabbitMQ 的 Exchange(交换器) 有4种类型，不同的类型对应着不同的路由策略：direct(默认)，fanout, topic, 和 headers，不同类型的Exchange转发消息的策略有所区别。

Queue

Queue(消息队列) 用来保存消息直到发送给消费者。RabbitMQ 中消息只能存储在队列中，它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

多个消费者可以订阅同一个队列，这时队列中的消息会被平均分摊（Round-Robin，即轮询）给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，这样避免的消息被重复消费。

工作模式

简单模式

RabbitMQ简单模式，也称为基本模式（Basic Model），是RabbitMQ的最简单的消息传递模式，仅涉及到一个生产者和一个消费者。

在这个模式中，当我们启动一个程序作为生产者并向RabbitMQ发出消息时，我们希望它直接进入队列中，然后消费者会从队列中获取这个消息并进行处理。

简单模式在RabbitMQ中是一个单队列单生产者单消费者的模式，主要适用于单纯的任务处理，消息的生产者和消费者的削峰填谷能力非常高。

工作队列模式

RabbitMQ工作队列模式，也称为Task Queues或Background Tasks，是一种常见的应用场景，它用于处理大量的任务，将任务进行排队，然后分发给多个消费者进行处理。这种模式适用于需要异步处理耗时的、密集型任务并且要求可靠性的情况。

RabbitMQ工作队列模式的基本原理是，将需要处理的任务投递到RabbitMQ中，生成任务队列（Task Queues），多个消费者通过拉取任务队列中的任务进行处理。

在RabbitMQ的工作队列模式中，队列中的每个消息都会分配给一个消费者进行处理。消费者可以是不同的进程、线程或服务，从而实现可扩展性和并行性。

在一个生产者-多个消费者的场景下，生产者只需要将消息发送到一个消息队列中，消费者会自动从队列中获取消息进行处理。如果存在多个消费者，队列中的消息将会被分摊给多个消费者进行处理，即实现了任务并行处理的功能。而且如果一个消费者挂掉，该消费者所占用的任务在一定的时间内不会被重新分配，即实现了任务可靠性处理的功能。

一般情况下，RabbitMQ的工作队列模式可以应用于以下场景：

任务比较繁重，处理较慢
任务多且耗时，无法同步处理
需要执行一些必须的后台任务，如发送邮件、生成报表等

发布订阅模式

RabbitMQ发布/订阅模式，也叫做“广播（Broadcast）模式”，是RabbitMQ的一种高级消息传递模式，主要用于广播消息。

在发布/订阅模式中，消息发送到Exchange（交换机）上，并携带着一个Routing Key（路由键），Exchange将收到的消息转发到绑定在它上面的所有队列。每个绑定键（Binding Key）都与一个队列相关联，而队列和消息的接收者实现了完全解耦，接收者只需要订阅（subscribe）与该队列相关联的绑定键即可。

我们将它作为“广播”模式，因为可以将一条消息同时发送到多个消费者。例如，我们可以让多个消费者接收网站上发布的新闻消息。

发布/订阅模式在RabbitMQ中的架构非常简单，主要可以描述为以下四个步骤：

生产者将消息发送到exchange中，并指定了Routing Key。
Exchange将消息分发到所有绑定它的队列上。
消费者从队列中接收消息，并进行处理。
消费者对队列进行确认操作，告诉RabbitMQ该消息已经被接收并处理。

路由模式

RabbitMQ路由模式是一种高级消息传递模式，它可以通过选择路由键（Routing Key）将消息推送到绑定键（Binding Key）与之匹配的队列中，以满足不同的消费者需要。

路由模式主要用于单一应用程序内的消息传递，生产者将消息发送到指定的Exchange（交换机）中，并且Exchange会根据Routing Key将消息放到绑定到Exchange上的队列中。而不同的消费者使用不同的Binding Key来决定与哪个队列建立联系并接收消息。

在RabbitMQ中，路由模式有以下几个步骤：

生产者将消息发送到Exchange中，并指定了Routing Key。
Exchange将消息根据Routing Key发送到绑定到Exchange中的队列。
消费者从队列中接收消息，并进行处理。
消费者对队列进行确认操作，告诉RabbitMQ该消息已经被接收并处理。

主题模式

RabbitMQ主题模式（Topic Model）是一种高级消息传递模式，它使你可以订阅一个特定的主题（Topic）并接收所有与该主题相关的消息。主题模式是在发布/订阅模式基础上进一步增强了消息传递的粒度。

在主题模式中，Exchange不仅可以使用Routing Key来将消息传递到队列中，还可以使用一个模式字符串来匹配Routing Key，这个模式字符串被称为主题（Topic）。消费者可以通过订阅不同的主题来接收不同的消息。

一个主题可以包含一个或多个单词（Word），单词之间使用"."（点号）来分割。通配符符号“#”表示跟单词数不限，而“”则表示只匹配一个单词。

主题模式在RabbitMQ中的架构非常简单，主要可以描述为以下四个步骤：

生产者将消息发送到Exchange中，并指定了Routing Key。
Exchange将消息根据匹配的主题字符串发送到绑定到Exchange中的队列。
消费者从队列中接收消息，并进行处理。
消费者对队列进行确认操作，告诉RabbitMQ该消息已经被接收并处理。

SpringBoot中使用RabbitMQ

在使用之前先要确保在电脑中已经安装了rabbitmq，并启动服务，启动成功后是可以在本地15672端口中访问到网页的。如图：

简单模式

基础配置

首先创建两个SpringBoot项目,一个作为消费者，一个作为生产者，并导入依赖项。

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

在两个项目的yaml文件中写入rabbitmq的配置信息

spring:
  rabbitmq:
    host: localhost # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: guest # 用户名
    password: guest # 密码

在rabbitmq中创建一个队列叫做queue1

生产者

在生产者中编写测试类，发送消息

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class RabbitPublisherApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void SimpleQueue(){
        //队列名字
        String name = "queue1";
        //具体的消息
        String message = "你好！！！rabbit";
        //发送消息
        rabbitTemplate.convertAndSend(name,message);
    }
}

执行测试类后在rabbitmq的相应队列中就会有消息了。

消费者

随后编写消费者来消费消息。

创建一个监听类，来监听具体的队列

@Component
public class SpringRabbitListener {
    /**
     * 从队列queue1中取消息
     * @param msg
     */
    @RabbitListener(queues = "queue1")
    public void listenSimpleQueueMessage(String msg) {
        System.out.println("spring 消费者接收到消息：" + msg);
    }
}

随后启动消费者项目，就可以从队列中成功获取到消息了。

工作队列模式

编写方法发送20条消息

    @Test
    public void WorkQueue(){
        String name = "work.queue";
        for(int i=0;i<20;i++){
            String message = "第"+i+"消息";
            rabbitTemplate.convertAndSend(name,message);
        }
    }

用两个消费者来接收

    @RabbitListener(queues = "work.queue")
    public void listenWorkQueueMessage1(String msg) {
        System.out.println("第一位消费者接收到消息：" + msg);
    }
    @RabbitListener(queues = "work.queue")
    public void listenWorkQueueMessage2(String msg) {
        System.out.println("第二位消费者接收到消息：" + msg);
    }

先启动消费者再发送消息：

此时可以发现两个消费者将消息均匀分配，每位各接收10条消息，这里是因为mq有预分配，会先预取，每个消费者和队列中有一个通道，存放预取的消息，所以会平均分消息，然后各自独立消费。

但是假如我们的消费者消费能力不一样呢，rabbitmq也会将其平均分配，这样的机制显然是不合理的，因此面对这种情况时，可以通过控制预取数量来改变这种情况，我们可以在yml文件中编写配置。这样就实现了取一个消费一个，不会先预取很多个，如果消费能力强取得就快，消费能力弱，取得就慢。

spring:
  rabbitmq:
    host: localhost # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: guest # 用户名
    password: guest # 密码
    listener:
      simple:
        prefetch: 1

发布/订阅模式（广播模式）

广播模式使用fanout交换机

首先编写一个配置类，配置交换机和队列

@Configuration
public class FanoutConfig {
    //声明交换机，设置名称
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("exchange1");
    }
    //队列1
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout1",true);
    }
    //绑定队列1和交换机
    @Bean
    public Binding binding1() {
        return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
    }
    //队列2
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout2",true);
    }
    //绑定交换机和队列2
    @Bean
    public Binding bindingQueue2() {
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    }
}

发送消息

    @Test
    public void fanout(){
        String name = "exchange1";
        String meg = "这里时fanout";
        rabbitTemplate.convertAndSend(name,"",meg);
    }

接收消息

    @RabbitListener(queues = "fanout1")
    public void listenWorkQueueMessage1(String msg) {
        System.out.println("第一位消费者接收到消息：" + msg);
    }
    @RabbitListener(queues = "fanout2")
    public void listenWorkQueueMessage2(String msg) {
        System.out.println("第二位消费者接收到消息：" + msg);
    }

结果我们可以看到两个消费者都接收到了消息。

路由模式

路由模式使用direct交换机,除了使用配置类的方式进行队列和交换机的绑定，还可以在消费者中使用注解的方式，例如

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name="direct",type = ExchangeTypes.DIRECT),
            key = {"one","two"}
    ))
    public void directListener1(String msg){
        System.out.println("第一位消费者接收到来自direct.queue1的消息："+ msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name="direct",type = ExchangeTypes.DIRECT),
            key = {"three"}
    ))
    public void directListener2(String msg){
        System.out.println("第二位消费者接收到来自direct.queue2的消息："+ msg);
    }

发送消息

    @Test
    public void directTest(){
        // 交换机名称
        String exchangeName = "direct";
        // 消息
        String message = "第 one 消息";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "one", message);

        // 消息
        String message2 = "第 two 消息";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "two", message2);

        // 消息
        String message3 = "第 three 消息";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "three", message3);
    }

最后可以看到，根据key路由接收到了消息。

主题模式

主题模式和路由模式类似，使用topic交换机，只是可以在key中使用通配符，

主题模式中RoutingKey一般由一个或多个单词组成，用“.”分割。

通配符规则

# 匹配一个或多个词

* 匹配一个词

修改消费者配置

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name="topic",type = ExchangeTypes.TOPIC),
            key = {"a.*"}
    ))
    public void topicListener1(String msg){
        System.out.println("第一位消费者接收到来自topic.queue1的消息："+ msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name="topic",type = ExchangeTypes.TOPIC),
            key = {"a.#"}
    ))
    public void topocListener2(String msg){
        System.out.println("第二位消费者接收到来自topic.queue2的消息："+ msg);
    }

发送消息

    @Test
    public void topicTest(){
        // 交换机名称
        String exchangeName = "topic";

        String message = "消息:a.b";
        rabbitTemplate.convertAndSend(exchangeName, "a.b", message);
        
        String message2 = "消息:a.b.c";
        rabbitTemplate.convertAndSend(exchangeName, "a.b.c", message2);

        String message3 = "消息:a.b.c.d";
        rabbitTemplate.convertAndSend(exchangeName, "a.b.c.d", message3);
    }

最后结果可以看到，第二个消费者可以拿到所有消息，第一个消费者只能拿到a.b的消息