6.4 工具-RabbitMQ

末蓝、 2024-03-30 14:38 10阅读 0赞

**目录**

6.4.1 RabbitMQ概述

6.4.1.1 什么是RabbitMQ

6.4.1.2 Erlang和AMQP

6.4.1.3 RabbitMQ

6.4.1.3.1 RabbitMQ优点

6.4.1.3.2 应用解耦

6.4.1.3.3 异步处理

6.4.1.3.4 流量削峰

6.4.2 RabbitMQ安装

6.4.3 RabbitMQ架构

6.4.3.1 架构图

6.4.3.1.1 术语

6.4.3.1.2 简单架构图

6.4.3.1.3 完整架构图

6.4.3.2 消息模式

6.4.4 SpringBoot整合RabbitMQ

6.4.4.1 简单模式

6.4.4.1.1 原生态方式

6.4.4.1.2 SpringBoot整合RabbitMQ方式

6.4.4.2 工作队列模式

6.4.4.2.1 说明

6.4.4.2.2 代码

6.4.4.3 发布订阅模式

6.4.4.3.1 说明

6.4.4.3.2 代码

6.4.4.4 路由模式

6.4.4.4.1 说明

6.4.4.4.2 代码

6.4.4.5 主题模式

6.4.4.5.1 说明

6.4.4.5.2 代码

6.4.5 RabbitMQ简单原理

6.4.5.1 Confirm机制

6.4.5.2 避免消息重复投递或重复消费

6.4.5.3 消息基于什么传输

6.4.5.4 消息分发

6.4.5.4.1 消息是如何分发的？

6.4.5.4.2 消息是如何路由的？

6.4.5.5 如何确保消息不丢失？

6.4.5.6 RabbitMQ的缺点

--------------------

## 6.4.1 RabbitMQ概述 ##

### 6.4.1.1 什么是RabbitMQ ###

假设有个注册动作，之后的流程如下图所示

![bee71e2ae91947daa6bab7d96ac5a2c7.png][]

耗时1.5有点长，能不能把后面的邮箱验证，短信验证在用户提交注册后，和添加数据库一起执行

### 6.4.1.2 Erlang和AMQP ###

Erlang是一门动态类型的函数式编程语言，它也是一门解释型语言，由Erlang虚拟机解释执行。从语言模型上说，Erlang是基于Actor模型的实现。在Actor模型里面，万物皆Actor，每个Actor都封装着内部状态，Actor相互之间只能通过消息传递这一种方式来进行通信。对应到Erlang里，每个Actor对应着一个Erlang进程，进程之间通过消息传递进行通信。相比共享内存，进程间通过消息传递来通信带来的直接好处就是消除了直接的锁开销(不考虑Erlang虚拟机底层实现中的锁应用)。

AMQP(Advanced Message Queue Protocol)定义了一种消息系统规范。这个规范描述了在一个分布式的系统中各个子系统如何通过消息交互。而RabbitMQ则是AMQP的一种基于erlang的实现。AMQP将分布式系统中各个子系统隔离开来，子系统之间不再有依赖。子系统仅依赖于消息。子系统不关心消息的发送者，也不关心消息的接受者。

### 6.4.1.3 RabbitMQ ###

#### 6.4.1.3.1 RabbitMQ优点 ####

常见的MQ：ActiveMQ，RocketMQ，Kafka，RabbitMQ。

*  语言的支持：ActiveMQ，RocketMQ只支持Java语言，Kafka可以支持多门语言，RabbitMQ支持多种语言。
 *  效率方面：ActiveMQ，RocketMQ，Kafka效率都是毫秒级别，RabbitMQ是微秒级别的。
 *  消息丢失，消息重复问题： RabbitMQ针对消息的持久化，和重复问题都有比较成熟的解决方案。
 *  学习成本：RabbitMQ非常简单。

RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控制、负载均衡等特性，使得RabbitMQ拥有更加广泛的应用场景。

**优点**

1.  应用解耦：降低系统模块的耦合度
2.  提高系统响应时间
3.  异步处理
4.  消息缓冲
5.  过载保护，流量削峰

#### 6.4.1.3.2 应用解耦 ####

用户下单后，订单系统需要通知库存系统，传统的做法就是订单系统调用库存系统的接口.

这种做法有一个缺点:

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

![9182fc86ee9545828e508227c529d2ce.png][]

**引入消息队列**

*  订单系统: 用户下单后，订单系统完成持久化处理，将消息写入消息队列,返回用户订单下单成功。
 *  库存系统: 订阅下单的消息,获取下单消息，进行库操作。 就算库存系统出现故障，消息队列也能保证消息的可靠投递，不会导致消息丢失

![ea39d06ef5ce47379f25468f355dbfed.png][]

#### 6.4.1.3.3 异步处理 ####

前面讲的用户注册案例

**1. 串行的方式**

![bee71e2ae91947daa6bab7d96ac5a2c7.png][]

**2. 并行的方式**

三个任务同时处理,返回给客户端,并行的方式能提高处理的时间

![5963fc2dca84425d9295f87ca2a64afd.png][]

#### 6.4.1.3.4 流量削峰 ####

因为流量过大，导致应用挂掉，为了解决这个问题，一般在应用前端加入消息队列。

## 6.4.2 RabbitMQ安装 ##

RabbitMQ在window和linux下均可安装。以下仅展示Linux下的RabbitMQ安装

**docker-compose.yml**

version: "3.1"
    services:
      rabbitmq:
        image: daocloud.io/library/rabbitmq:management
        restart: always
        container_name: rabbitmq
        ports:
          - 5672:5672 # 服务器端口
          - 15672:15672 # 客户端访问服务器的端口
        volumes:
          - ./data:/var/lib/rabbitmq

[root@iz8vbdmrir2n6xqzrbd93hz /]# cd /opt
    # 创建文件夹
    [root@iz8vbdmrir2n6xqzrbd93hz opt]# mkdir docker_rabbitmq
    # 进入文件夹
    [root@iz8vbdmrir2n6xqzrbd93hz opt]# cd docker_rabbitmq/
    # 添加docker-compose.yml 将上面的配置信息粘贴进来
    [root@iz8vbdmrir2n6xqzrbd93hz docker_rabbitmq]# vim docker-compose.yml
    [root@iz8vbdmrir2n6xqzrbd93hz docker_rabbitmq]# docker-compose-Linux-x86_64 up -d

![8a6f18cf67bf440bb52d614b808bcdab.png][]

# 查看所有进程
    [root@192 docker_rabbitmq]# docker ps -a

![d1e101559d18466985c22f316c9ab0af.png][]

> 浏览器访问：http://**自己的ip**:15672
> 
> 云服务器记得开放 15672 和 5672 端口
> 
> 用户名和密码默认都是：guest

访问

![62ebf0e0a1984947a2ad7a1d1e8c905d.png][]

**RabbitMQ的图形化管理界面**

![ec23328e341a445e95d574345c60c17c.png][]

切换到：docker_rabbitmq 下
    # down 停止服务并删除
    docker-compose-Linux-x86_64 down 
    # down 停止服务
    docker-compose-Linux-x86_64 stop     
    systemctl stop firewalld
    systemctl disable firewalld
    systemctl restart docker 
    # 启动服务
    docker-compose-Linux-x86_64 up -d

## 6.4.3 RabbitMQ架构 ##

### 6.4.3.1 架构图 ###

#### 6.4.3.1.1 术语 ####

*  Publisher - 生产者：发布消息到RabbitMQ中的Exchange
 *  Consumer - 消费者：监听RabbitMQ中的Queue中的消息
 *  Exchange - 交换机：和生产者建立连接并接收生产者的消息
 *  Queue - 队列：Exchange会将消息分发到指定的Queue，Queue和消费者进行交互
 *  Routes - 路由：交换机以什么样的策略将消息发布到Queue

#### 6.4.3.1.2 **简单架构图**  ####

![1aae9cb30c81476fbfd8809595674ecd.png][]

Publisher生产消息，将消息放到Exchange上，根据Routes将消息分发到Queue，Consumer处理消息

#### 6.4.3.1.3 **完整架构图** ####

![16781717f6f4425eaf22282500f9855d.png][]

生产者和虚拟机建立连接，建立完连接后创建一个Channel,通过Channel把消息发送到Exchange上

### 6.4.3.2 消息模式 ###

官网【[RabbitMQ Tutorials — RabbitMQ][RabbitMQ Tutorials _ RabbitMQ]】

常见的五种模式

1.  简单模式
2.  Work Queues 工作队列模式
3.  Publish/Subscribe发布订阅模式
4.  Routing 路由模式
5.  Topics 主题模式

###  ###

## 6.4.4 SpringBoot整合RabbitMQ ##

### 6.4.4.1 简单模式 ###

#### 6.4.4.1.1 原生态方式 ####

原生态方式实现简单五种消息模型中的**Simple-简单模型。**

由于所有的功能都要自己手动引入依赖，手动写代码，所以会比较繁琐

以后用springboot整合后会更简便，springboot会优化代码，同时隐藏一些代码

利用SpringBoot自动生成项目结构

如果不知道如何自动生成点击【[6.2 微服务-SpringBoot\_老李头喽的博客-CSDN博客][6.2 _-SpringBoot_-CSDN]】

导入依赖

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

之后开始创建生产者，消费者

![de09168f78564fe4880243797dd77f36.png][]

**创建Publisher发布者**

//生产者
    public class Publisher {
        public static void main(String[] args) throws Exception{
    
            /*
            * 1.创建连接工厂
            * 2.获取连接对象
            * 3.创建频道
            * 4.声明队列
            * 5.发布消息
            * 6.释放资源
            */
    
            System.out.println("Publisher...");
            //1.创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //配置连接参数
            factory.setHost("192.168.72.129");
            //连接服务器端口5672，而不是在浏览器客户端访问端口15672
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            //2.获取连接对象
            Connection connection = factory.newConnection();
            //3.创建频道
            Channel channel = connection.createChannel();
            //4.声明队列
            //配置队列参数
            //参数1：queue - 指定队列的名称
            //参数2：durable - 当前队列是否需要持久化,值为true时表示持久化，rabbitmq宕机或重启后，队列依然在
            //参数3：exclusive - 当前队列是否为排他队列，值为true时表示与当前连接（connection）绑定，连接关闭，队列消失
            //参数4：autoDelete - 当前队列是否自动删除，值为true时表示队列中的消息一旦被消费，该队列会消失
            //参数5：arguments - 指定当前队列的相关参数
            channel.queueDeclare("helloworldQueue",false,false,false,null);
            //5.发布消息
            //发布消息到exchange，同时指定路由的规则
            // 参数1：指定exchange，使用""，简单模式没有交换机
            // 参数2：指定路由的规则，使用具体的队列名称
            // 参数3：指定传递的消息所携带的properties，使用null
            // 参数4：指定发布的具体消息，byte[]类型
            channel.basicPublish("","helloworldQueue",null,"发布者发布，helloworld".getBytes());
            //6.释放资源
            //注意顺序
            // channel.close();
            // connection.close();
        }
    }

**创建Consumer消费者**

//消费者
    public class Consumer {
        public static void main(String[] args)throws Exception {
            System.out.println("Consumer...");
            //配置连接参数
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.72.129");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            //获取连接
            Connection connection = factory.newConnection();
    
            Channel channel = connection.createChannel();
    		//要与生产者中的该方法一致（注：方法中的参数值必须保持一致）
            channel.queueDeclare("helloworldQueue",false,false,false,null);
    
            // 回调函数：消费(处理业务)的过程
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
                //重写DefaultConsumer里的handleDelivery方法
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者接收到消息："+new String(body));
                }
            };
    
            //消费消息
            //参数1：queue - 指定消费哪个队列
            //参数2：autoAck - 指定是否自动ACK （true表示接收到消息后，会立即告知RabbitMQ，false表示不告知）
            //参数3：consumer - 指定消费回调函数
            channel.basicConsume("helloworldQueue",true,defaultConsumer);
        }
    }

启动Docker

![0331165760df454c869157de2bae8be3.png][]

分别启动生产者和消费者进行测试

> 运行生产者，Queues中会有一条消息，当运行消费者后，消息会被处理，Queues中消息为0

由于没有设置持久化，服务器宕机后，消息队列会清空

> 在声明队列时需要配置队列参数  
> //参数1：queue - 指定队列的名称  
> //参数2：durable - 当前队列是否需要持久化,值为true时表示持久化，rabbitmq宕机或重启后，队列依然在  
> //参数3：exclusive - 当前队列是否为排他队列，值为true时表示与当前连接（connection）绑定，连接关闭后队列也会随着消失  
> //参数4：autoDelete - 当前队列是否自动删除，值为true时表示队列中的消息一旦被消费，该队列会消失  
> //参数5：arguments - 指定当前队列的相关参数

#### 6.4.4.1.2 SpringBoot整合RabbitMQ方式 ####

为了区别上面的原生态方式，新建包

![dd41671ac0cb4e43b15f08e477ed81fb.png][]

**引入启动器**

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

**application.properties中增加配置**

# 应用名称
    spring.application.name=rabbitmq
    # 应用服务 WEB 访问端口
    server.port=8080
    
    #rabbitMQ配置
    spring.rabbitmq.host=192.168.72.129
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

**SimpleConfig**

//配置类
    @Configuration
    public class SimpleConfig {
        @Bean
        public Queue getQueue() {
            return new Queue("simpleQueue", true, false, false, null);
        }
    }

注意Queue包

![e995cc08f49049d4ab5aa5b3e22dc674.png][]

**Publisher**

//生产者
    @Component
    public class Publisher {
        @Autowired
        RabbitTemplate rabbitTemplate;
        public void send(){
            System.out.println("生产者：发布消息");
            rabbitTemplate.convertAndSend("","simpleQueue","消息：有一个消息");
        }
    }

**Consumer**

//消费者
    @Component
    @RabbitListener(queues = "simpleQueue")  //时刻监听着指定的队列
    public class Consumer {
        @RabbitHandler //@RabbitListener监听消息队列中是否有消息，有消息就调用@RabbitHandler标识的方法处理
        public void receive(String msg) {
            System.out.println("接收到消息:" + msg);
        }
    }

消费者需要用到@RabbitListener和@RabbitHandler注解

![007729fb9c5b42b3ba66e0bf0de8e3a8.png][]

**测试类**

@RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitmqApplication2 {
        @Autowired
        Publisher publisher;
        @Test
        public  void test(){
            publisher.send();
        }
    }

由于消费者有监听机制，所以只需要运行测试类里的publisher中的send方法即可，当消息队列中有消息，监听机制会自动使消费者处理消息

> 注意如果报以下空指针错误
> 
> java.lang.NullPointerException  
> at com.bl.RabbitmqApplication2.test(RabbitmqApplication2.java:18)
> 
> 检查测试类是否添加了以下两个注解
> 
> @RunWith(SpringRunner.class)  
> @SpringBootTest
> 
> 原因
> 
> Springboot的@RunWith(SpringRunner.class)
> 
> 注解的意义在于Test测试类要使用注入的类，比如@Autowired注入的类，
> 
> 有了@RunWith(SpringRunner.class)这些类才能实例化到spring容器中，自动注入才能生效，
> 
> 否则会抛出NullPointerExecption  
>

### 6.4.4.2 工作队列模式 ###

#### 6.4.4.2.1 说明 ####

![524566ff3c884c9f99c5d65adefda110.png][]

工作队列模式不需要交换机

一个生产者，一个队列，多个消费者，默认采用公平分发

#### 6.4.4.2.2 代码 ####

**WorkConfig配置类**

@Configuration
    public class WorkConfig {
        //工作模式：一个队列，多个消费者
        @Bean
        public Queue workQueue(){
            System.out.println("workqueue模式创建队列");
            return new Queue("workQueue",true);
        }
    }

**Publisher生产者**

**Consumer消费者1**

//消费者
    @Component
    @RabbitListener(queues = "workQueue")  //时刻监听着指定的队列
    public class Consumer {
        @RabbitHandler //@RabbitListener监听消息队列中是否有消息，有消息就调用@RabbitHandler标识的方法处理
        public void receive(String msg) {
            System.out.println("Consumer1 接收到消息:" + msg);
        }
    }

**Consumer2 消费者2**

//消费者
    @Component
    @RabbitListener(queues = "workQueue")  //时刻监听着指定的队列
    public class Consumer2 {
        @RabbitHandler //@RabbitListener监听消息队列中是否有消息，有消息就调用@RabbitHandler标识的方法处理
        public void receive(String msg) {
            System.out.println("Consumer2 接收到消息:" + msg);
        }
    }

### 6.4.4.3 发布订阅模式 ###

#### 6.4.4.3.1 说明 ####

![82673565d66348c8a760256763d36dbd.png][]

使用该模式需要借助交换机，生产者将消息发送到交换机，再通过交换机到达队列。

交换机分四种：

*  Direct：直接交换机，通过消息上的路由键(Routing key)直接对消息进行分发，相当于精确匹配，一对一
 *  Topic：交换机会将路由键和绑定上的模式进行通配符匹配，相当于模糊匹配，一对多
 *  Headers：消息头交换机，使用消息头的属性进行消息路由，相当于模糊匹配，一对多
 *  Fanout：扇出交换机，会将消息发送到所有和它进行绑定的队列上，广播，群发

**默认交换机是Direct，发布订阅模式使用Fanout类型交换器**

#### 6.4.4.3.2 代码 ####

**FanoutConfig**

@Configuration
    public class FanoutConfig {
        //发布订阅模式  一个生产者   一个交换机(FanoutExchange)  两个队列（队列需要绑定到交换机）  两个消费者
        //创建扇出交换机
        @Bean
        public FanoutExchange fanoutExchange(){
            System.out.println("创建交换机");
            return new FanoutExchange("fanoutExchange",true,false);
        }
        //创建队列
        @Bean
        public Queue fanoutqueue1(){
            return new Queue("fanoutqueue1",true);
        }
        @Bean
        public Queue fanoutqueue2(){
            return new Queue("fanoutqueue2",true);
        }
        //将队列绑定到交换机
        //通过方法获得对象
        @Bean
        public Binding bindingqueque1(){
        //通过BindingBuilder的bind方法将队列和交换机绑定在一起
          return  BindingBuilder.bind(fanoutqueue1()).to(fanoutExchange());
        }
        @Bean
        public Binding bindingqueque2(){
            return  BindingBuilder.bind(fanoutqueue2()).to(fanoutExchange());
        }
        //通过形参获得对象
    
    }

//生产者
    @Component
    public class Publisher {
    
            @Autowired
            RabbitTemplate rabbitTemplate;
            public void send(){
                System.out.println("生产者：发布消息");
                rabbitTemplate.convertAndSend("fanoutExchange","","生产者：发布订阅模式");
            }
        }

//消费者
    @Component
    @RabbitListener(queues = "fanoutqueue1")  //时刻监听着指定的队列
    public class Consumer1 {
        @RabbitHandler //@RabbitListener监听消息队列中是否有消息，有消息就调用@RabbitHandler标识的方法处理
        public void receive(String msg) {
            System.out.println("Consumer1接收到消息:" + msg);
        }
    }

//消费者
    @Component
    @RabbitListener(queues = "fanoutqueue2")  //时刻监听着指定的队列
    public class Consumer2 {
        @RabbitHandler //@RabbitListener监听消息队列中是否有消息，有消息就调用@RabbitHandler标识的方法处理
        public void receive(String msg) {
            System.out.println("Consumer2接收到消息:" + msg);
        }
    }

### 6.4.4.4 路由模式 ###

#### 6.4.4.4.1 说明 ####

![c3e23aabbda84dbba132f0cf22260ada.png][]

一个生产者，一个交换机，多个队列，多个消费者

生产者将消息发送到Direct交换机（路由模式需要借助直连交换机实现），在绑定队列和交换机的时候有一个路由键(Routing key)，生产者发送的消息会指定一个路由键，那么消息只会发送到相应key相同的队列，接着监听该队列的消费者消费消息。消费者可以选择性的接收消息

#### 6.4.4.4.2 代码 ####

**RoutingConfig**

//路由模式 一个生产者 一个交换机（直连交换机） 两个队列（路由key） 两个消费者
    @Configuration
    public class RoutingConfig {
        //创建直连交换机
        @Bean
        public DirectExchange directExchange(){
            return new DirectExchange("directExchange");
        }
        //队列
        @Bean
        public Queue routingQueue1(){
            return new Queue("routingQueue1");
        }
        @Bean
        public Queue routingQueue2(){
            return new Queue("routingQueue2");
        }
        //绑定
        @Bean
        public Binding bingroutingQueue1(){
            //直连交换机必须带routingkey
            return BindingBuilder.bind(routingQueue1()).to(directExchange()).with("1");
        }
        @Bean
        public Binding bingroutingQueue2(){
            //直连交换机必须带routingkey
            return BindingBuilder.bind(routingQueue2()).to(directExchange()).with("2");
        }
    }

**注意： 路由模式，绑定队列和交换机时需要指定路由键**

**RoutingPublisher**

//生产者
    @Component
    public class RoutingPublisher {
    
            @Autowired
            RabbitTemplate rabbitTemplate;
            public void send(){
                    System.out.println("生产者：发布消息");
                //设置routingkey
                rabbitTemplate.convertAndSend("directExchange","1","生产者：消息 1");
                rabbitTemplate.convertAndSend("directExchange","2","生产者：消息 2");
                rabbitTemplate.convertAndSend("directExchange","3","生产者：消息 3");
            }
        }

//消费者
    @Component
    @RabbitListener(queues = "routingQueue1")  //时刻监听着指定的队列
    public class RoutingConsumer1 {
        @RabbitHandler //@RabbitListener监听消息队列中是否有消息，有消息就调用@RabbitHandler标识的方法处理
        public void receive(String msg) {
            System.out.println("Consumer1接收到消息:" + msg);
        }
    }

//消费者
    @Component
    @RabbitListener(queues = "routingQueue2")  //时刻监听着指定的队列
    public class RoutingConsumer2 {
        @RabbitHandler //@RabbitListener监听消息队列中是否有消息，有消息就调用@RabbitHandler标识的方法处理
        public void receive(String msg) {
            System.out.println("Consumer2接收到消息:" + msg);
        }
    }

### 6.4.4.5 主题模式 ###

#### 6.4.4.5.1 说明 ####

![4b9f4c27e7c6466995591cba22d274f5.png][]

一个生产者，一个交换机，多个队列，多个消费者

主题模式又称通配符模式

使用直连交换机可以改善我们的系统，但是它仍有局限性，它不能实现**多重条件**的路由。

在消息系统中，我们不仅想要订阅基于路由键的队列，还想订阅基于生产消息的源。这时候可以使用Topic交换机。

使用主题交换机时不能采用任意写法的路由键，路由键的形式应该是由点分割的有意义的单词。例如"goods.stock.info"等。路由key最多255字节。

*  \* ：代表一个单词
 *  \# ：代表0个或多个单词

#### 6.4.4.5.2 代码 ####

@Configuration
    public class TopicConfig {
    
        @Bean
        public Queue topicQueue1(){
            return new Queue("topicQueue1");
        }
    
        @Bean
        public Queue topicQueue2(){
            return new Queue("topicQueue2");
        }
    
        @Bean
        public TopicExchange topicExchange(){
            return new TopicExchange("topicExchange");
        }
    
        @Bean
        public Binding bingTopicQueue1(){
            return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("China.*");
        }
    
        @Bean
        public Binding bingTopicQueue2(){
            return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("America.#");
        }
    }

@Component
    public class TopicPublisher {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Autowired
        private TopicExchange topicExchange;
    
        public void send() {
            System.out.println("TopicPublisher");
            rabbitTemplate.convertAndSend(topicExchange.getName(), "China.beijing", "中国北京");
            rabbitTemplate.convertAndSend(topicExchange.getName(), "America.Texas.Houston", "美国德克萨斯休斯顿");
        }
    }

@Component
    @RabbitListener(queues = "topicQueue1")
    public class TopicCustomer1 {
    
        @RabbitHandler
        public void receive(String msg) {
            System.out.println("TopicCustomer1：" + msg);
        }
    }

@Component
    @RabbitListener(queues = "topicQueue2")
    public class TopicCustomer2 {
    
        @RabbitHandler
        public void receive(String msg) {
            System.out.println("TopicCustomer2：" + msg);
        }
    }

![fb25919799124747a7cccf08aabe3bc4.png][]

但是如果将 TopicPublisher 中修改为

rabbitTemplate.convertAndSend(topicExchange.getName(), "China.beijing.haidian", "中国北京海淀区");

结果 TopicCustomer1就接收不到消息中国北京海淀

![7d3480fa47d74800b529b4ca7467e5e7.png][]

原因就是

*  \* ：代表一个单词
 *  \# ：代表0个或多个单词

## 6.4.5 RabbitMQ简单原理 ##

### 6.4.5.1 Confirm机制 ###

**发送方确认模式**

将信道设置成 confirm 模式（发送方确认模式），则所有在信道上发布的消息都会被指派一个唯一的 ID。 一旦消息被投递到目的队列后，或者消息被写入磁盘后（可持久化的消息），信道会发送一个确认给生产者（包含消息唯一 ID）。 如果 RabbitMQ 发生内部错误从而导致消息丢失，会发送一条 nack（not acknowledged，未确认）消息。

发送方确认模式是异步的，生产者应用程序在等待确认的同时，可以继续发送消息。当确认消息到达生产者应用程序，生产者应用程序的回调方法就会被触发来处理确认消息。

**接收方确认机制**

消费者接收每一条消息后都必须进行确认（消息接收和消息确认是两个不同操作）。只有消费者确认了消息，RabbitMQ 才能安全地把消息从队列中删除。 这里并没有用到超时机制，RabbitMQ 仅通过 Consumer 的连接中断来确认是否需要重新发送消息。也就是说，只要连接不中断，RabbitMQ 给了 Consumer 足够长的时间来处理消息。保证数据的最终一致性；

如何避免消息重复投递或重复消费？

### 6.4.5.2 避免消息重复投递或重复消费 ###

在消息生产时，MQ 内部针对每条生产者发送的消息生成一个 inner-msg-id，作为去重的依据（消息投递失败并重传），避免重复的消息进入队列； 在消息消费时，要求消息体中必须要有一个 bizId（对于同一业务全局唯一，如支付 ID、订单 ID、帖子 ID 等）作为去重的依据，避免同一条消息被重复消费。

### 6.4.5.3 消息基于什么传输 ###

由于 TCP 连接的创建和销毁开销较大，且并发数受系统资源限制，会造成性能瓶颈。RabbitMQ 使用信道的方式来传输数据。信道是建立在真实的 TCP 连接内的 虚拟连接，且每条 TCP 连接上的信道数量没有限制。

### 6.4.5.4 消息分发 ###

#### 6.4.5.4.1 消息是如何分发的？ ####

若该队列至少有一个消费者订阅，消息将以循环（round-robin）的方式发送给消费者。每条消息只会分发给一个订阅的消费者（前提是消费者能够正常处理消息并进行确认）。

通过路由可实现多消费的功能

#### 6.4.5.4.2 消息是如何路由的？ ####

消息提供方 → 路由 → 一或多个队列

消息发布到交换器时，消息将拥有一个路由键（routing key），在消息创建时设定。 通过队列路由键，可以把队列绑定到交换器上。 消息到达交换器后，RabbitMQ 会将消息的路由键与队列的路由键进行匹配（针对不同的交换器有不同的路由规则）

交换机分四种：

*  Direct：直接交换机，通过消息上的路由键(Routing key)直接对消息进行分发，相当于精确匹配，一对一
 *  Topic：交换机会将路由键和绑定的模式进行通配符匹配，相当于模糊匹配，一对多
 *  Headers：消息头交换机，使用消息头的属性进行消息路由，相当于模糊匹配，一对多
 *  Fanout：扇出交换机，会将消息发送到所有和它进行绑定的队列上，广播，群发

### 6.4.5.5 如何确保消息不丢失？ ###

消息持久化，当然前提是队列必须持久化

RabbitMQ 确保持久性消息能从服务器重启中恢复的方式是，将它们写入磁盘 的一个持久化日志文件，当发布一条持久性消息到持久交换器上时，RabbitMQ 会在消息提交到日志文件后才发送响应。 一旦消费者从持久队列中消费了一条持久化消息，RabbitMQ 会在持久化日志中把这条消息标记为等待垃圾收集。如果持久化消息在被消费之前 RabbitMQ 重启， 那么 Rabbit 会自动重建交换器和队列（以及绑定），并重新发布持久化日志文件 中的消息到合适的队列。

### 6.4.5.6 RabbitMQ的缺点 ###

**系统可用性降低**

系统引入的外部依赖越多，越容易挂掉，本来你就是 A 系统调用 BCD 三个系统的接口就好了，人 家ABCD 四个系统好好的，没啥问题，偏加个 RabbitMQ 进来，万一 RabbitMQ 挂了咋整？RabbitMQ 挂了，整套系统崩溃了。

**系统复杂性提高**

硬生生加个RabbitMQ 进来，怎么保证消息没有重复消费？怎么处理消息丢失的情况？ 怎么保证消息传递的顺序性？问题一大堆，痛苦不已

**一致性问题**

A 系统处理完了直接返回成功了，人都以为你这个请求就成功了；但是问题是，要是 BCD 三个系统那里，BD 两个系统写库成功了，结果 C 系统写库失败了，咋整？ 你这数据就不一致了。 所以消息队列实际是一种非常复杂的架构，引入它有很多好处，但是也得针对它带来的坏处做各种额外的技术方案和架构来规避掉，最后，系统复杂度提升了一个数量级（也许是复杂了 10 倍）。但是关键时刻还是得用。

[bee71e2ae91947daa6bab7d96ac5a2c7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/547cc90c5db74e4e88bf04979ff4b748.png
[9182fc86ee9545828e508227c529d2ce.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/130848d1cbc44efbb218e507568c6819.png
[ea39d06ef5ce47379f25468f355dbfed.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/22f353d7e77d4ec8b2f662f0f6ef3b9a.png
[5963fc2dca84425d9295f87ca2a64afd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/f3d78c77093249188e4bfe0c32611f32.png
[8a6f18cf67bf440bb52d614b808bcdab.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/45ea6812f46249d3a33406c744ece40c.png
[d1e101559d18466985c22f316c9ab0af.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/6ead7e4ec39a44d58a6aabc1440377f9.png
[62ebf0e0a1984947a2ad7a1d1e8c905d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/6f851764968442bba89b557a0ba07b24.png
[ec23328e341a445e95d574345c60c17c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/985f15739bd0482db7e30e7aedb70ed1.png
[1aae9cb30c81476fbfd8809595674ecd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/b2e0876534fb449bb4c3d10823a8a774.png
[16781717f6f4425eaf22282500f9855d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/64a485b6d45e45799204ae24a3f8de49.png
[RabbitMQ Tutorials _ RabbitMQ]: https://www.rabbitmq.com/getstarted.html
[6.2 _-SpringBoot_-CSDN]: https://blog.csdn.net/qq_43757965/article/details/128472023?spm=1001.2014.3001.5502
[de09168f78564fe4880243797dd77f36.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/1b98469d87a84d4fbffaf8baa2e09afd.png
[0331165760df454c869157de2bae8be3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/dc0686a3fe2340a68b8a8221c5fb16d8.png
[dd41671ac0cb4e43b15f08e477ed81fb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/7dca6baeb13f4d958bcbc569405bd922.png
[e995cc08f49049d4ab5aa5b3e22dc674.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/c2454ef6a73641058858845f05b5582e.png
[007729fb9c5b42b3ba66e0bf0de8e3a8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/527f20343941432d8e5064e544d66e28.png
[524566ff3c884c9f99c5d65adefda110.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/b4f5266fc7c74a8b886657db4ad10fb4.png
[82673565d66348c8a760256763d36dbd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/360420ba9c8949f09a73f081c40eec52.png
[c3e23aabbda84dbba132f0cf22260ada.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/7b8ed07f16134765bd906927b357bca0.png
[4b9f4c27e7c6466995591cba22d274f5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/d220cdccba484ed18a5d3d7e0c948733.png
[fb25919799124747a7cccf08aabe3bc4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/5f30220e878b4e9188b5265891ca667b.png
[7d3480fa47d74800b529b4ca7467e5e7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/ac4ceb9fda7040f59ad7dd2cf0c444bd.png