RabbitMQ概念与管理端配置说明

环境信息

操作系统：Windows 10
RabbitMQ 版本：4.1.1

安装教程

一.RabbitMQ概念说明

        RabbitMQ 是一个基于 AMQP（Advanced Message Queuing Protocol，高级消息队列协议） 的开源消息中间件，最初由 LShift 公司开发，后由 VMware 收购并维护。它采用 Erlang 编写，以其高可靠性、灵活性和可扩展性著称。RabbitMQ 通过消息队列实现分布式系统中的异步通信、削峰解耦和负载均衡。下面是其消息传递架构图，说明了RabbitMQ的一些核心组件以及交互流程。

消息传递架构图

• Producer：消息的发布者；
• Consumer：消息的订阅者；
• Broker：RabbitMQ的实例，只有一个；
• Connection：与Broker之间的TCP连接；
• Channel：是基于 Connection 创建的轻量级逻辑通道，用于操作某个Vhost下的资源；
• Vhost：虚拟主机，逻辑隔离单元，一个Broker可以有多个Vhost；
• Exchange：交换机，可以理解为路由规则。有多种不同的交换机，区别在于路由规则的不同；
• Binding：Exchange和Queue之间的映射规则，确定哪些Queue接收待交换机的信息。
• Queue：消息对接

二.RabbitMQ的配置

       通过上面的消息传递架构图，可以清除的看到RabbitMQ的各个组件，那么这些组件如何创建和配置呢？

2.1Connection

        由客户端创建(如Java)，下面是管理端的详解。下面这张图，显示的是连接RabbitMQ的Connection.

下面每个字段的解释：

2.1Channel

在客户端连接Rabbit MQ时创建。下面这张图，显示的是连接打开的Channel。

 下面每个字段的解释：

2.3Producer

public static void main(String[] args) throws IOException, TimeoutException {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");  // RabbitMQ 地址factory.setPort(5672);        // 默认端口factory.setUsername("guest"); // 默认用户名factory.setPassword("guest"); // 默认密码factory.setVirtualHost("/");  // 默认虚拟主机// 2. 建立连接和通道try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 3. 声明队列（如果已存在则不会重复创建）/*** 参数说明：* 1. 队列名称* 2. 是否持久化* 3. 是否独占队列* 4. 是否自动删除* 5. 额外的参数*/channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 4. 发送消息String message = "Hello from plain Java!";channel.basicPublish("", QUEUE_NAME, null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}}

2.4Consumer

 public static void main(String[] args) throws Exception {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);factory.setUsername("guest");factory.setPassword("guest");factory.setVirtualHost("/");// 2. 建立连接和通道Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3. 声明队列（确保队列存在）channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 4. 创建消费者对象DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};// 5. 订阅队列channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});}

2.5Vhost

2.5.1管理界面创建

创建成功

字段说明：

• Name：为新的 Vhost 指定一个唯一的名称；
• Description：提供 Vhost 的描述信息，用于说明该 Vhost 的用途或相关细节。
• Tags：为 Vhost 添加标签，便于管理和分类。
•  Default Queue Type：选择该 Vhost 中创建队列时的默认队列类型。
  • Classic（经典队列）：传统的、广泛使用的队列类型，适用于大多数场景。
  • Quorum（仲裁队列）：提供更强的一致性和容错能力，适用于需要高可用性的场景。
  • Stream（流队列）：这是一种针对大数据流设计的队列类型，适用于处理大量连续的数据流。

• Overview：提供 Vhost 的基本信息和状态概览。
  • Name（名称）：Vhost 的名称。
  • Users（用户）：可以访问该 Vhost 的用户列表。
  • State（状态）：Vhost 的当前运行状态（如 running 表示正在运行）。
• Messages: 显示与该 Vhost 相关的消息统计信息。
  • Ready：队列中已准备好被消费的消息数量。
  • Unacked：已被消费者获取但尚未确认的消息数量。
  • Total：队列中的总消息数量（包括 Ready 和 Unacked）。
• Network：展示与该 Vhost 相关的网络流量情况。
  • From client：从客户端到 RabbitMQ 的数据传输速率（单位：字节/秒）。
  • To client：从 RabbitMQ 到客户端的数据传输速率（单位：字节/秒）。
• Message rates：显示与该 Vhost 相关的消息发布和交付速率。
  • publish：每秒发布的消息数量。
  • deliver / get：每秒交付或获取的消息数量。

2.5.2命令行创建

D:\cmd>rabbitmqctl add_vhost /dev
Adding vhost "/dev" ...D:\cmd>rabbitmqctl list_vhosts
Listing vhosts ...
name
/
/devD:\cmd>

2.6Exchange

        在创建Vhost的时候，默认会创建7个Exchange。通常不需要手动创建Exchange，默认的Exchange就包含了所有的类型。

2.6.1管理页面创建

• Virtual host（虚拟主机）：表示该交换机所属的虚拟主机（Vhost）。
• Name（名称）：交换机的名称。
• Type（类型）：交换机的类型，决定了消息如何路由到队列。
  • direct：完全匹配键名进行路由。
  • fanout：广播模式，将消息发送给所有绑定的队列。
  • topic：基于通配符匹配进行路由。
  • headers：基于消息头属性进行路由。
• Durability（持久性）：设置交换机是否持久化。
  • Durable：持久化，服务器重启后交换机依然存在。
  • Transient：非持久化，服务器重启后交换机会被删除。
• Auto delete（自动删除）：设置当所有队列与该交换机解绑时，是否自动删除该交换机。
  • Yes：自动删除。
  • No：不自动删除。
• Internal（内部）：设置该交换机是否为内部交换机。
  • Yes：内部交换机，只能通过其他交换机进行路由。
  • No：普通交换机，可以直接发布消息。
• Arguments（参数）：设置交换机的额外参数。
• Alternate exchange（备用交换机）：当消息无法根据当前交换机规则路由到任何队列时，可以将其转发到指定的备用交换机进行处理。

已有交换机列表

• Features（特性）：交换机支持的特性。
  • D：表示该交换机是持久化的（Durable），即在服务器重启后仍然存在。
  • I：表示该交换机是内部的（Internal），只能通过其他交换机进行路由，不能直接发布消息。
• Message rate in（消息输入速率）：单位时间内进入该交换机的消息数量。
• Message rate out（消息输出速率）：单位时间内从该交换机路由到队列的消息数量。 

 2.6.2.命令行创建

       通过命令行创建Exchange，需要 rabbitmqadmin.exe插件，RabbitMQ默认不带这个插件，通过http://localhost:15672/cli/rabbitmqadmin下载，是的，通过本地安装的RabbitMQ服务下载，所以要保证它在运行，windows系统会下载rabbitmqadmin.exe，放入sbin文件夹下即可，linux或macOS会下载rabbitmqadmin(是一个python文件)。也有识别出现问题的时候，比如我，在windwos系统下载了个python文件，当然也可以把它打包成exe，或者别的地方下载。但我不想弄了，有机会再说。

2.7Binding

        在创建新的Vhost后，RabbitMQ默认生成了7个Exchange，除了(AMQP default)，其他都要配置Binding(如果使用的话)。(AMQP default)默认Binding当前Vhost的所有消息队列。配置Binding的步骤如下：

下面是添加binding的页面以及一些介绍

• To queue: 将当前Exchange绑定到一个Queue,输入Queue的名称
• To Exchange: 将当前交换机绑定到另一个交换机
• Routing key：设置 路由键，不是必填项，但通常需要根据交换机类型来设置。

• Arguments：设置绑定时的 额外参数（Arguments）
• String（下拉框）：选择 Arguments 中值的 数据类型。

2.8Queue

与Exchange不同，Queue需要手动创建，下面是创建Queue的一些说明

• Add a new queue:
  • Virtual host：选择要创建 Queue 的 虚拟主机（Vhost）
  •  Type：选择 Queue 的 类型。
  • Name：输入 Queue 的 名称，可以自定义，但需符合 RabbitMQ 的命名规范（通常为小写字母、数字和特殊字符 -_.）。
  • Durability：设置 Queue 的 持久化属性。
    • Durable（持久化）：服务器重启后 Queue 依然存在；
    • Transient（非持久化）：服务器重启后 Queue 会被删除。
  • Arguments：设置 Queue 的 额外参数（Arguments）

• 其他可选配置项
  • Auto expire：设置队列的自动过期时间（毫秒），如果队列在指定时间内没有被使用（即没有任何消费者连接），则自动删除。
  • Message TTL：设置队列中消息的过期时间（毫秒），超过这个时间的消息会被自动丢弃或转移到死信队列（如果有配置）。
  • Overflow behaviour：设置队列溢出时的行为。
    • drop-head：丢弃队首消息；
    • reject-publish：拒绝发布新消息（返回给生产者）；
    • reject-publish-dlx：拒绝发布新消息并发送到死信队列（如果有配置）。
  • Single active consumer：启用单活动消费者模式。在同一时刻，只有一个消费者可以消费队列中的消息，其他消费者处于等待状态。
  • Dead letter exchange / routing key：设置死信交换机及其路由键。当消息过期、被拒绝或队列达到最大长度时，将消息转发到指定的死信交换机进行进一步处理。
  •  Max length / Max length bytes：设置队列的最大长度（消息数）或最大字节数。限制队列的大小，防止无限增长。
  • Leader locator：仅对 Quorum 队列有效，设置 Leader 节点的选择策略。
    • client-local：优先选择本地节点作为 Leader；
    • least-index：选择索引最小的节点作为 Leader

上面这些可配置选项，就是添加参数，点击后会自动创建一个Arguments行，值需手动填写。