3、RabbitMQ的入门和消息分发机制

Navy2026-08-04mqrabbitmq

前面已经介绍了Rabbit MQ的安装部署和上手初体验，可以感受到Rabbit MQ的使用还是比较简单的，特别是在SpringBoot或SpringCloud环境下都可以非常丝滑地集成到我们的项目中，接下来会介绍几个Rabbit MQ相关的几个基础知识

1、基础补充

1.1、插件

Rabbit MQ提供了非常丰富的插件，插件目录在根目录的plugins目录下，前面的内容也介绍到可以通过下面命令启用一个插件，比如要启用延迟消息插件，只需要先将这个插件下载下来并且放到plugins目录下，然后执行以下rabbitmq-plugins enable xxx

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

如果我们是用docker或者k8s运行的Rabbit MQ，那么大概率已经自动启用了UI管理界面，可以通过15672端口访问。如果插件没有启用的化也可以手动启用

rabbitmq-plugins enable rabbitmq_management

1.2、用户管理

Rabbit MQ默认的账号和密码是guest，如果我们要添加自己的用户，可以直接在UI管理后台Admin菜单下的Users子菜单页面进行添加，当然也可以直接通过Rabbit MQ提供的命令行工具添加

添加用户

rabbitmqctl add_user admin admin

添加一个admin账号，密码也是admin

为用户添加操作权限

rabbitmqctl set_user_tags admin administrator

给新添加的admin账号设置成administrator权限

为用户添加资源权限

rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

第一个.*（配置权限）：允许admin配置所有资源（比如创建、删除交换机/队列等操作）
第二个.*（读权限）：允许admin读取所有资源（比如消费消息、查询队列信息等）
第三个.*（写权限）：允许admin写入所有资源（比如发送消息到交换机/队列）

所以上面命令行的意思就是授权admin对vhost为"/"的配置权限、读权限、写权限

1.3、端口号说明

4369：是Erlang的端口/结点名称映射程序，用来跟踪节点名称监听，在集群中起到类似DNS的作用

5672、5671：AMQP客户端端口，没有使用ssl和使用ssl的端口

25672：用于Rabbit MQ节点之间和CLI工具通信，配合4369端口

15672：HTTP API端口，管理员用户才能访问，用于管理Rabbit MQ，需要启用Management插件

61613、61614：当STOMP插件启用的时候才打开，作为STOMP客户端端口（根据是否使用TLS选择）

1883、8883：当mqtt插件启用时打开，作为mqtt客户端端口（根据是否使用TLS选择）

15674：基于websocket的STOMP客户端端口（当插件web stomp启用的时候打开）

15675：基于websocket的mqtt客户端端口（当启用web mqtt启用的时候打开）

1.4、基本概念

下图是Rabbit MQ的业务模型

1.4.4、channel

和Rabbit MQ的通信管道，用于真正发送消息或者接收消息，可以简单理解为和Rabbit MQ的链接，不同点是同一个connection可能会维护多个channel

1.4.6、vhost

虚拟主机，主要起到资源隔离的作用，类似Linux的cgoup或者k8s里边的namespace。在我的工作用一般习惯用来区分环境

1.4.7、Exchanges

交换机，主要负责消息的分发逻辑。如果发送消息的时候不指定交换机名称，程序会自动去找当前vhost下的默认交换机

1.4.9、路由键

生产者在往交换机上发送消息的时候，主要指定路由键，交换机会根据这个路由键去匹配消息的分发规则

1.4.10、绑定键

队列和交换机的绑定关系，在消息分发的时候会根据路由键来匹配这个绑定键，如果满足对应交换机类型的匹配规则，则消息就会被路由到对应队列上

1.4.11、消息

消息一般主要由两部分组成：

消息体：

消息体一般是带有具体业务逻辑的数据结构，比如json格式（可以进一步格式化或者序列化）

附加信息：

附加信息是消息的额外表述，比如目标交换机的名称、路由键以及一些其它的自定义属性等，可以根据自己的业务自由扩展

2、消息的分发机制

消息的分发规则主要跟交换机的类型有关系，而默认的交换机有四种类型：direct（直连类型交换机）、topic（主题类型交换机）、fanout（广播类型交换机）、headers（基于消息头类型交换机）

2.1、Direct（直连交换机）

直连类型交换机的特点是交换机和队列的绑定键是一个固定的值，不支持通配符。而且也只有发送消息时的消息路由键和绑定键完全相同的时候消息才会被分发到绑定的队列上

如图，比如有一个直连类型的交换机，一个队列和这个交换机的绑定键是order，另一个队列和这个交换机的绑定键是user。如果这个时候往这个交换机上发送了一条消息，如果发送消息时指定的路由键是order，那么只有第一个队列会收到这条消息，如果发送消息的时候指定的路由键是user，则只有第二个队列会收到这条消息

2.2、Topic（主题类型交换机）

主题类型交换机的绑定键支持通配符"*"和"#"，区别在于

"*"：代表一个单词，英文点号"."分割

"#"：代表一个或多个单词，英文点号"."分割

比如一个topic类型的交换机和一个队列的绑定键是order.#，如果生产者往这个交换机发送一条消息。路由键是order.toy.create，则这个队列会收到这个消息。如果交换机和队列的绑定键是order.toy.*，那么这个队列也会收到这条消息。而如果队列和交换机的绑定键是order.*，那么发送消息的时候如果指定路由键为order.toy.create，则这个队列将无法收到这条消息

所以在使用的时候一定要结合自己的业务情况考虑清楚该如何设计绑定键和路由键

2.3、fanout（广播类型交换机）

广播类型交换机的特点是“广播”，故名思意就是一条消息发送到交换机上，绑定到这个交换机的所有队列都会收到这条消息。而且交换机和队列之间可以无需设置绑定键，发送消息的时候也无需设置路由键

2.4、headers（消息头类型交换机）

消息的分发和发送消息的时候设置的header信息有关系，可以通过设置自定义k-v匹配规则来进行Arguments匹配，发送消息的时候需要指定header，并且header里的值必须要和绑定键Arguments设置的值相匹配，比如：

创建绑定关系：

@Bean
public Binding bindingA(HeadersExchange headersExchange,Queue queueA){
    HashMap<String, Object> map = new HashMap<>();
    map.put("type","OK");
    map.put("status","200");
    return BindingBuilder.bind(queueA).to(headersExchange).whereAll(map).match();
}

发送消息：

public void sendMsg(){
   //消息属性
    MessageProperties messageProperties=new MessageProperties();
    HashMap<String, Object> map = new HashMap<>();
    map.put("type","error");
    map.put("status","500");
    //设置消息头
    messageProperties.setHeaders(map);
    //添加消息属性
    Message message = MessageBuilder.withBody("hello word!".getBytes(StandardCharsets.UTF_8))
            .andProperties(messageProperties).build();
    rabbitTemplate.convertAndSend("exchange.headers","",message);
}

header类型的交换机一般在工作中不怎么用，性能也比较差，它不是根据路由键和绑定键来进行消息分发，而是根据发送的消息的headers属性进行匹配。

消息的流转过程

1、生产者连接到Rabbit MQ的Broker，建立Connection链接，并开启一个信道Channel

2、生产者声明一个交换机，并设置相关的属性，比如交换机的类型、是否持久化等

3、生产者声明一个队列，并设置相关的属性，比如是否持久化、是否自动删除等

4、消费者确认（ack）接收到消息

5、Rabbit MQ从队列中删除相应已经被确认消费的消息

6、关闭信道、关闭链接

扫码关注