1. RabbitMQ 基础

文档摘要

RabbitMQ 基础 RabbitMQ 基础详解：构建可靠消息传递系统消息队列与 RabbitMQ 概述 1.1 什么是消息队列？消息队列是一种在应用程序之间传递消息的异步通信机制。它充当消息的中间人，允许生产者（Producer）将消息发送到队列中，而消费者（Consumer）则从队列中接收和处理消息。这种模式实现了生产者和消费者之间的解耦，带来了诸多优势：解耦: 生产者和消费者无需直接了解彼此，只需关注消息本身，降低系统耦合度。异步: 生产者发送消息后无需等待消费者响应，可以立即返回，提高系统响应速度。削峰填谷: 消息队列可以缓冲突发流量，平滑系统负载，防止系统崩溃。可靠性: 消息队列通常具备消息持久化和确认机制，确保消息的可靠传递。

1. RabbitMQ 基础

RabbitMQ 基础详解：构建可靠消息传递系统

1. 消息队列与 RabbitMQ 概述

1.1 什么是消息队列？

消息队列是一种在应用程序之间传递消息的异步通信机制。它充当消息的中间人，允许生产者（Producer）将消息发送到队列中，而消费者（Consumer）则从队列中接收和处理消息。这种模式实现了生产者和消费者之间的解耦，带来了诸多优势：

解耦: 生产者和消费者无需直接了解彼此，只需关注消息本身，降低系统耦合度。
异步: 生产者发送消息后无需等待消费者响应，可以立即返回，提高系统响应速度。
削峰填谷: 消息队列可以缓冲突发流量，平滑系统负载，防止系统崩溃。
可靠性: 消息队列通常具备消息持久化和确认机制，确保消息的可靠传递。
可伸缩性: 可以通过增加队列和消费者实例来扩展系统处理能力。

1.2 RabbitMQ 简介

RabbitMQ 是一个开源的消息代理软件，最初基于 AMQP（Advanced Message Queuing Protocol，高级消息队列协议）协议实现，后来也扩展支持 STOMP、MQTT 等多种协议。它使用 Erlang 语言开发，具有高并发、高可用、易扩展等特点。RabbitMQ 的核心目标是提供可靠的消息传递服务，帮助构建松耦合、可伸缩的分布式系统。

RabbitMQ 的关键特性:

可靠性 (Reliability): 提供多种机制确保消息的可靠传递，如消息持久化、消息确认等。
灵活路由 (Flexible Routing): 支持多种消息路由策略，满足不同的业务需求。
消息集群 (Clustering): 支持集群部署，提高系统可用性和吞吐量。
多协议支持 (Multi-protocol Support): 除了 AMQP，还支持 STOMP、MQTT 等协议，方便不同场景的应用集成。
易用性 (Ease of Use): 提供友好的管理界面和丰富的客户端库，方便开发和管理。

1.3 RabbitMQ 应用场景

RabbitMQ 在各种场景下都有广泛的应用，例如：

异步任务处理: 将耗时任务放入消息队列，由消费者异步处理，提高 Web 应用响应速度。
服务解耦: 微服务架构中，使用 RabbitMQ 解耦服务之间的依赖关系，提高系统灵活性。
事件驱动架构: 基于 RabbitMQ 构建事件驱动系统，实现服务之间的松耦合和实时响应。
日志收集: 将日志消息发送到 RabbitMQ，进行统一处理和分析。
消息通知: 实现用户注册、订单通知等异步消息通知功能。

2. RabbitMQ 核心概念

为了更好地理解 RabbitMQ 的工作原理，我们需要了解其核心概念。以下是 RabbitMQ 中最重要的几个组件：

2.1 生产者 (Producer)

生产者是消息的发送者，负责创建和发送消息到 RabbitMQ 服务器。生产者可以是任何应用程序，例如 Web 应用、移动应用或后台服务。

2.2 消费者 (Consumer)

消费者是消息的接收者，负责从 RabbitMQ 服务器接收消息并进行处理。消费者通常是后台服务或应用程序，它们持续监听队列并处理到达的消息。

2.3 消息代理 (Broker)

RabbitMQ Broker 是消息队列服务器的核心组件，负责接收、存储和路由消息。Broker 主要由以下几个关键部分组成：

交换机 (Exchange): 接收生产者发送的消息，并根据路由规则将消息路由到一个或多个队列。
队列 (Queue): 存储消息，等待消费者消费。
绑定 (Binding): 定义交换机和队列之间的关联关系，指定消息如何从交换机路由到队列。
路由键 (Routing Key): 生产者在发送消息时指定，交换机根据路由键和绑定规则来路由消息。
虚拟主机 (Virtual Host，vhost): 提供逻辑隔离，允许在同一个 RabbitMQ Broker 上创建多个独立的虚拟 Broker 环境。
信道 (Channel): 客户端与 RabbitMQ Broker 建立连接后，通过信道进行通信，一个连接可以创建多个信道。

2.4 交换机 (Exchange)

交换机是消息的入口点，它接收生产者发送的消息，并根据交换机类型和绑定规则将消息路由到一个或多个队列。RabbitMQ 提供了四种常用的交换机类型：

Direct Exchange (直连交换机): 将消息路由到 Routing Key 完全匹配 的队列。

graph LR
Producer --> Exchange[Direct Exchange]
Exchange -- Routing Key route_key_A --> QueueA
Exchange -- Routing Key route_key_B --> QueueB
QueueA --> ConsumerA
QueueB --> ConsumerB



* **Fanout Exchange (扇形交换机/广播交换机):** 将消息 **广播** 到所有绑定到该交换机的队列，忽略 Routing Key。

    ```mermaid
    graph LR
        Producer --> Exchange(Fanout Exchange)
        Exchange --> QueueA
        Exchange --> QueueB
        QueueA --> ConsumerA
        QueueB --> ConsumerB
    ```

* **Topic Exchange (主题交换机):**  使用 **Routing Key 模式匹配** 的方式将消息路由到一个或多个队列。可以使用 `*` 和 `#` 通配符进行模式匹配。

    * `*`: 匹配一个单词。

    * `#`: 匹配零个或多个单词。

    ```mermaid

graph LR
    Producer --> Exchange[Topic Exchange]
    Exchange -- Routing Key log.info --> QueueInfo
    Exchange -- Routing Key log.error.db --> QueueError
    Exchange -- Routing Key order.create --> QueueOrder
    QueueInfo --> ConsumerInfo
    QueueError --> ConsumerError
    QueueOrder --> ConsumerOrder

Headers Exchange (头部交换机): 根据消息的 headers 属性 进行路由，而不是 Routing Key。相比 Topic Exchange 更加灵活，但性能稍差。

2.5 队列 (Queue)

队列是消息的存储容器，用于存储等待消费者处理的消息。队列具有以下特点：

FIFO (First-In, First-Out): 消息按照先进先出的顺序存储和消费。
持久化 (Durability): 队列可以被声明为持久化的，即使 RabbitMQ 服务重启，队列仍然存在。
独占性 (Exclusive): 队列可以被声明为独占的，只允许一个消费者连接到该队列。
自动删除 (Auto-delete): 队列可以被声明为自动删除的，当最后一个消费者断开连接后，队列会被自动删除。

2.6 绑定 (Binding)

绑定定义了交换机和队列之间的关联关系。它指定了交换机如何将消息路由到队列。绑定通常包含以下信息：

交换机名称: 绑定的交换机。
队列名称: 绑定的队列。
路由键 (Routing Key): 用于 Direct Exchange 和 Topic Exchange 的路由规则。
Headers (可选): 用于 Headers Exchange 的头部属性匹配规则。

2.7 路由键 (Routing Key)

路由键是生产者在发送消息时指定的一个字符串，用于指示消息的路由目标。交换机根据路由键和绑定规则来决定将消息路由到哪些队列。

2.8 虚拟主机 (Virtual Host)

虚拟主机 (vhost) 提供了逻辑隔离，允许在同一个 RabbitMQ Broker 上创建多个独立的虚拟 Broker 环境。每个 vhost 拥有独立的交换机、队列、绑定和用户权限，不同 vhost 之间的资源相互隔离。这使得在同一个 RabbitMQ Broker 上可以安全地运行多个应用程序或团队的服务。

2.9 信道 (Channel)

信道是客户端与 RabbitMQ Broker 建立连接后，用于进行消息通信的通道。一个 TCP 连接可以创建多个信道，每个信道都可以执行不同的 RabbitMQ 操作，例如声明交换机、队列、绑定、发送和接收消息等。使用信道可以有效地复用 TCP 连接，减少资源消耗，提高系统性能。

3. RabbitMQ 消息流转过程

了解 RabbitMQ 的消息流转过程有助于深入理解其工作原理。一个典型的消息流转过程如下：

生产者 (Producer) 连接到 RabbitMQ Broker，并创建一个信道 (Channel)。
生产者声明交换机 (Exchange)，指定交换机类型和名称。 (如果交换机不存在)
生产者声明队列 (Queue)，指定队列属性和名称。 (如果队列不存在)
生产者创建绑定 (Binding)，将交换机和队列绑定起来，并指定路由键 (Routing Key)。 (如果绑定不存在)
生产者发送消息到交换机，并指定 Routing Key。
交换机根据交换机类型和绑定规则，将消息路由到一个或多个队列。
消息被存储在队列中，等待消费者消费。
消费者 (Consumer) 连接到 RabbitMQ Broker，并创建一个信道 (Channel)。
消费者声明要消费的队列。
消费者订阅队列，开始接收消息。
RabbitMQ Broker 将队列中的消息推送给消费者。
消费者接收到消息后进行处理，并发送确认 (Acknowledgement) 给 RabbitMQ Broker。
RabbitMQ Broker 收到确认后，从队列中删除消息。

4. RabbitMQ 代码实践 (Python + Pika)

接下来，我们将通过 Python 语言和 Pika 客户端库，演示 RabbitMQ 的基本操作，包括生产者发送消息和消费者接收消息。

4.1 环境准备

首先，确保您已经安装了 RabbitMQ 服务器并运行。您可以使用 Docker 快速启动 RabbitMQ：


docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

5672: AMQP 协议端口
15672: RabbitMQ Management UI 端口 (用户名/密码: guest/guest)

然后，安装 Pika 客户端库：


pip install pika

4.2 生产者代码 (producer.py)


#!/usr/bin/env python
import pika
# 连接 RabbitMQ 服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# 声明交换机 (Direct Exchange)
exchange_name = 'direct_logs'
channel.exchange_declare(exchange=exchange_name, exchange_type='direct')
# 定义路由键 (severity)
severities = ['info', 'warning', 'error']
for severity in severities:
    message = f"This is a {severity} log message."
    # 发布消息到交换机，指定路由键
    channel.basic_publish(exchange=exchange_name,
                          routing_key=severity,
                          body=message)
    print(f" [x] Sent {severity}: {message}")
# 关闭连接
connection.close()

代码详解:

pika.BlockingConnection(pika.ConnectionParameters('localhost')): 创建与 RabbitMQ 服务器的阻塞连接。ConnectionParameters('localhost') 指定连接到本地 RabbitMQ 服务器。
connection.channel(): 创建一个信道。
channel.exchange_declare(exchange=exchange_name, exchange_type='direct'): 声明一个名为 direct_logs 的 Direct Exchange。如果交换机已存在，则不会重复创建。
severities = ['info', 'warning', 'error']: 定义路由键列表。
channel.basic_publish(...): 发布消息到交换机。
- exchange=exchange_name: 指定交换机名称。
- routing_key=severity: 指定路由键，循环遍历 severities 列表。
- body=message: 消息内容。
connection.close(): 关闭连接。

4.3 消费者代码 (consumer.py)


#!/usr/bin/env python
import pika
# 连接 RabbitMQ 服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# 声明交换机 (Direct Exchange)，与生产者保持一致
exchange_name = 'direct_logs'
channel.exchange_declare(exchange=exchange_name, exchange_type='direct')
# 声明队列 (随机队列名，exclusive=True 表示独占队列，connection 关闭后自动删除)
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
# 获取需要监听的路由键 (从命令行参数获取，默认监听所有)
import sys
severities = sys.argv[1:]
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
    sys.exit(1)
# 绑定队列到交换机，根据路由键进行绑定
for severity in severities:
    channel.queue_bind(exchange=exchange_name,
                       queue=queue_name,
                       routing_key=severity)
    print(f" [*] Binding queue '{queue_name}' to exchange '{exchange_name}' with routing key '{severity}'")
def callback(ch, method, properties, body):
    print(f" [x] Received {method.routing_key}: {body.decode()}")
    # 手动确认消息 (确保消息被处理后才删除)
    ch.basic_ack(delivery_tag=method.delivery_tag)
# 设置消息消费者，并指定回调函数
channel.basic_consume(queue=queue_name,
                      on_message_callback=callback,
                      auto_ack=False) # 关闭自动确认，使用手动确认
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

代码详解:

前几行代码与生产者代码类似，建立连接、创建信道、声明交换机。
channel.queue_declare(queue='', exclusive=True): 声明一个匿名队列 (服务器随机生成队列名)，exclusive=True 表示队列是独占的，只允许当前连接访问，连接关闭后自动删除。
queue_name = result.method.queue: 获取服务器生成的队列名。
severities = sys.argv[1:]: 从命令行参数获取需要监听的路由键。例如，运行 python consumer.py info error 将只接收路由键为 info 和 error 的消息。
channel.queue_bind(...): 将队列绑定到交换机，并指定路由键。消费者可以绑定多个路由键，接收符合任何一个路由键的消息。
def callback(ch, method, properties, body):: 定义消息回调函数，当收到消息时会被调用。
- method.routing_key: 获取消息的路由键。
- body.decode(): 将消息体 (bytes 类型) 解码为字符串。
- ch.basic_ack(delivery_tag=method.delivery_tag): 手动确认消息。
channel.basic_consume(...): 设置消息消费者。
- queue=queue_name: 指定要消费的队列。
- on_message_callback=callback: 指定消息回调函数。
- auto_ack=False: 关闭自动确认，启用手动确认。
channel.start_consuming(): 开始消费消息，进入阻塞状态，等待接收消息。

4.4 运行示例

启动消费者: 在终端中运行消费者代码，并指定要监听的路由键，例如：
```
python consumer.py info error
```
启动生产者: 在另一个终端中运行生产者代码：
```
python producer.py
```

观察结果: 消费者终端会输出接收到的消息，例如：


 [*] Binding queue 'amq.ctag-...' to exchange 'direct_logs' with routing key 'info'
 [*] Binding queue 'amq.ctag-...' to exchange 'direct_logs' with routing key 'error'
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received info: b'This is a info log message.'
 [x] Received error: b'This is a error log message.'

生产者终端会输出发送的消息：


 [x] Sent info: This is a info log message.
 [x] Sent warning: This is a warning log message.
 [x] Sent error: This is a error log message.

可以看到，消费者只接收到了路由键为 info 和 error 的消息，而 warning 消息被交换机丢弃，因为没有队列绑定了 warning 路由键。

5. 总结与展望

本文详细介绍了 RabbitMQ 的基础知识，包括消息队列的概念、RabbitMQ 的核心组件、消息流转过程以及代码实践。通过学习本文，您应该对 RabbitMQ 有了初步的了解，并能够使用 Pika 客户端库进行简单的消息发布和消费。

总结:

RabbitMQ 是一个强大的开源消息代理软件，适用于构建可靠的分布式系统。
核心概念包括生产者、消费者、Broker、交换机、队列、绑定、路由键、虚拟主机和信道。
交换机类型决定了消息的路由策略，常用的有 Direct、Fanout、Topic 和 Headers 交换机。
代码实践演示了如何使用 Python 和 Pika 客户端库进行消息的发布和消费。

展望:

RabbitMQ 的功能远不止于此，还有很多高级特性值得深入学习，例如：

消息持久化 (Message Persistence): 确保消息在 RabbitMQ 服务重启后不会丢失。
消息确认 (Message Acknowledgement): 确保消息被消费者成功处理。
消息 TTL (Time-To-Live): 设置消息的过期时间。
死信队列 (Dead Letter Exchange, DLX): 处理无法被正常消费的消息。
优先级队列 (Priority Queue): 支持消息优先级。
RabbitMQ 集群 (Clustering): 提高系统可用性和吞吐量。

希望本文能帮助您入门 RabbitMQ，并激发您进一步学习和探索 RabbitMQ 的热情。在实际应用中，请根据您的业务需求选择合适的 RabbitMQ 特性和配置，构建高效、可靠的消息传递系统。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： https://aiknowledge.cn/collection/216-rabbitmq%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97%E5%AE%9E%E6%88%98%E6%95%99%E7%A8%8B

整理：灏天文库整理

由灏天文库平台收录，内容或由平台用户上传，仅供学习交流