2. RabbitMQ 核心概念详解

文档摘要

RabbitMQ 核心概念详解 RabbitMQ 核心概念详解：从基础领域到实践应用 RabbitMQ 基础领域：消息队列的价值与应用场景在现代分布式系统中，服务之间的通信变得日益复杂。为了应对高并发、解耦服务、提高系统可用性等挑战，消息队列（Message Queue，简称 MQ）应运而生。RabbitMQ 作为一款开源的消息代理（Message Broker），在众多 MQ 产品中脱颖而出，成为构建可靠、可扩展的分布式系统的关键组件。 1.1 消息队列的核心价值：异步处理（Asynchronous Processing）：消息队列允许生产者（Producer）将消息发送到队列中，无需等待消费者（Consumer）立即处理。消费者可以稍后从队列中取出消息进行处理。

2. RabbitMQ 核心概念详解

RabbitMQ 核心概念详解：从基础领域到实践应用

1. RabbitMQ 基础领域：消息队列的价值与应用场景

在现代分布式系统中，服务之间的通信变得日益复杂。为了应对高并发、解耦服务、提高系统可用性等挑战，消息队列（Message Queue，简称 MQ）应运而生。RabbitMQ 作为一款开源的消息代理（Message Broker），在众多 MQ 产品中脱颖而出，成为构建可靠、可扩展的分布式系统的关键组件。

1.1 消息队列的核心价值：

异步处理（Asynchronous Processing）： 消息队列允许生产者（Producer）将消息发送到队列中，无需等待消费者（Consumer）立即处理。消费者可以稍后从队列中取出消息进行处理。这种异步处理模式极大地提高了系统的响应速度和吞吐量。例如，用户注册场景，用户注册成功后，发送欢迎邮件、短信通知等非核心业务可以放入消息队列异步处理，避免阻塞主流程。
服务解耦（Service Decoupling）： 消息队列充当服务之间的中间层，生产者和消费者之间无需直接依赖。生产者只需要将消息发送到消息队列，而无需关心哪个消费者会处理它。消费者也只需要从消息队列中获取消息，无需知道消息的来源。这种解耦降低了服务之间的耦合度，提高了系统的可维护性和可扩展性。例如，订单系统和库存系统可以通过消息队列进行通信，订单系统下单后，发送消息到消息队列，库存系统订阅消息进行库存扣减，两个系统可以独立部署和升级。
流量削峰（Traffic Shaping/Buffering）： 在高并发场景下，消息队列可以作为缓冲层，将短时间内涌入的大量请求放入队列中，然后消费者按照自身处理能力逐步消费队列中的消息。这样可以有效地缓解后端服务的压力，防止系统崩溃。例如，秒杀活动中，瞬间涌入大量的下单请求，消息队列可以缓存这些请求，后端服务按照处理能力慢慢处理，避免系统被瞬间流量冲垮。
可靠消息传递（Reliable Message Delivery）： RabbitMQ 提供了多种机制来保证消息的可靠传递，例如消息持久化、消息确认机制等。即使在网络故障或 Broker 宕机的情况下，消息也不会丢失，确保数据的一致性。这对于金融交易、订单处理等对数据可靠性要求极高的场景至关重要。
广播与路由（Publish/Subscribe & Routing）： 消息队列支持发布/订阅模式和路由模式，允许生产者将消息发送给多个消费者，或者根据消息的路由键将消息发送给特定的消费者。这为构建复杂的事件驱动架构提供了强大的支持。

1.2 RabbitMQ 的典型应用场景：

微服务架构： 在微服务架构中，服务之间需要频繁地进行通信。RabbitMQ 可以作为微服务之间的消息总线，实现服务间的异步通信和解耦。
异步任务处理： 对于耗时较长的任务，例如图片处理、视频转码、数据分析等，可以将任务放入消息队列，由后台消费者异步处理，提高用户体验。
日志收集： 可以将分布在不同服务器上的日志信息发送到消息队列，然后由日志分析系统统一收集和分析。
事件驱动架构： 在事件驱动架构中，系统中的各个组件通过发布和订阅事件进行交互。RabbitMQ 可以作为事件总线，实现事件的发布、路由和订阅。
物联网（IoT）： 在物联网场景中，大量的设备需要将数据上传到云端。RabbitMQ 可以作为数据接入层，接收设备上传的数据，并进行后续处理。

2. RabbitMQ 核心概念详解：构建消息传递的基石

理解 RabbitMQ 的核心概念是掌握其使用和构建高效消息系统的关键。下面我们将逐一深入解析 RabbitMQ 的核心概念，并结合代码实践和 Mermaid 图表进行说明。

2.1 Message（消息）：

消息是 RabbitMQ 中最基本的数据单元，它是在生产者和消费者之间传递的信息载体。一条消息通常包含两个主要部分：

Payload（消息体）： 消息的实际内容，可以是任何格式的数据，例如文本、JSON、二进制数据等。
Properties（消息属性）： 描述消息的元数据，例如消息的类型、优先级、持久性、过期时间等。Properties 可以帮助消费者更好地处理消息，也可以影响消息在 RabbitMQ Broker 中的路由和存储行为。

代码实践（Python - pika）：


import pika
# 连接 RabbitMQ Broker
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# 声明 Exchange (稍后详细介绍)
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
# 消息体
message_body = "Hello, RabbitMQ!"
# 消息属性
properties = pika.BasicProperties(
    delivery_mode=2,  # 消息持久化
    content_type='text/plain', # 消息内容类型
)
# 发布消息到 Exchange
channel.basic_publish(exchange='direct_logs',
                      routing_key='info',  # 路由键 (稍后详细介绍)
                      body=message_body,
                      properties=properties)
print(f" [x] Sent '{message_body}'")
connection.close()

代码详解：

pika.BasicProperties 用于创建消息属性对象。
delivery_mode=2 设置消息为持久化，即使 RabbitMQ Broker 重启，消息也不会丢失。
content_type='text/plain' 指定消息内容的类型为纯文本。
channel.basic_publish 方法用于发布消息，需要指定 Exchange 名称、路由键、消息体和消息属性。

2.2 Exchange（交换机）：

Exchange 是 RabbitMQ 的核心组件之一，它负责接收生产者发送的消息，并根据预先定义的规则将消息路由到一个或多个 Queue（队列）。Exchange 本身不存储消息，它只负责消息的路由。

RabbitMQ 提供了四种常用的 Exchange 类型，每种类型根据不同的路由策略将消息路由到队列：

Direct Exchange（直连交换机）：

Direct Exchange 根据消息的 Routing Key（路由键） 将消息路由到 Binding Key（绑定键） 完全匹配的 Queue。 Binding Key 是 Queue 在绑定 Exchange 时指定的。

Mermaid 图示：

graph TD
Producer --> Exchange[Direct Exchange]
Exchange --> |Routing Key = route_key_A| QueueA[Queue A]
Exchange --> |Routing Key = route_key_B| QueueB[Queue B]
QueueA --> Consumer1[Consumer 1]
QueueB --> Consumer2[Consumer 2]
style Exchange fill:#f9f,stroke:#333,stroke-width:2px
style QueueA fill:#ccf,stroke:#333,stroke-width:2px
style QueueB fill:#ccf,stroke:#333,stroke-width:2px





    **代码实践（Python - pika）：**



    **生产者 (producer.py):**



    ```python
    import pika
    import sys
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
    severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
    message = ' '.join(sys.argv[2:]) or 'Hello World!'
    channel.basic_publish(exchange='direct_logs',
                          routing_key=severity,  # 使用 severity 作为 Routing Key
                          body=message.encode())
    print(f" [x] Sent {severity}:{message}")
    connection.close()
    ```



    **消费者 (consumer.py):**



    ```python
    import pika
    import sys
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
    result = channel.queue_declare(queue='', exclusive=True) # 声明匿名队列
    queue_name = result.method.queue
    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='direct_logs',
                           queue=queue_name,
                           routing_key=severity) # 绑定队列到 Exchange，指定 Binding Key
    print(' [*] Waiting for logs. To exit press CTRL+C')
    def callback(ch, method, properties, body):
        print(f" [x] {method.routing_key}:{body.decode()}")
    channel.basic_consume(queue=queue_name,
                          on_message_callback=callback,
                          auto_ack=True)
    channel.start_consuming()
    ```



    **代码详解：**



    * **生产者：**

        * 使用 `channel.exchange_declare(exchange='direct_logs', exchange_type='direct')` 声明一个 Direct Exchange，名称为 `direct_logs`。

        * 使用 `channel.basic_publish` 发布消息，`routing_key` 参数的值决定了消息的路由目标。

    * **消费者：**

        * 使用 `channel.queue_declare(queue='', exclusive=True)` 声明一个匿名队列（队列名称由 RabbitMQ 自动生成），`exclusive=True` 表示队列是排他队列，只会被声明它的连接使用，连接断开后队列自动删除。

        * 使用 `channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity)` 将队列绑定到 `direct_logs` Exchange，并指定 `routing_key` (Binding Key)。  消费者可以绑定多个 Binding Key，接收不同 Routing Key 的消息。

        * 消费者接收消息后，根据 `method.routing_key` 可以知道消息的 Routing Key。



* **Fanout Exchange（扇形交换机）：**



    Fanout Exchange 将接收到的所有消息 **广播** 到所有绑定到该 Exchange 的 Queue，忽略 Routing Key。  类似于广播，所有订阅者都会收到消息。



    **Mermaid 图示：**



    ```mermaid
    graph TD
        Producer --> Exchange(Fanout Exchange)
        Exchange --> QueueA(Queue A)
        Exchange --> QueueB(Queue B)
        QueueA --> Consumer1(Consumer 1)
        QueueB --> Consumer2(Consumer 2)
        style Exchange fill:#f9f,stroke:#333,stroke-width:2px
        style QueueA fill:#ccf,stroke:#333,stroke-width:2px
        style QueueB fill:#ccf,stroke:#333,stroke-width:2px
    ```



    **代码实践（Python - pika）：**



    **生产者 (emit_log.py):**



    ```python
    import pika
    import sys
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs', exchange_type='fanout') # 声明 Fanout Exchange
    message = ' '.join(sys.argv[1:]) or "info: Hello World!"
    channel.basic_publish(exchange='logs', routing_key='', body=message.encode()) # routing_key 忽略
    print(f" [x] Sent {message}")
    connection.close()
    ```



    **消费者 (receive_logs.py):**



    ```python
    import pika
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs', exchange_type='fanout') # 声明 Fanout Exchange
    result = channel.queue_declare(queue='', exclusive=True) # 声明匿名队列
    queue_name = result.method.queue
    channel.queue_bind(exchange='logs', queue=queue_name) # 绑定队列到 Exchange，无需 Routing Key
    print(' [*] Waiting for logs. To exit press CTRL+C')
    def callback(ch, method, properties, body):
        print(f" [x] {body.decode()}")
    channel.basic_consume(queue=queue_name,
                          on_message_callback=callback,
                          auto_ack=True)
    channel.start_consuming()
    ```



    **代码详解：**



    * **生产者和消费者：** 都声明了 `fanout` 类型的 Exchange，名称为 `logs`。

    * **生产者：**  `channel.basic_publish` 中的 `routing_key` 参数被忽略，因为 Fanout Exchange 不关心 Routing Key。

    * **消费者：**  `channel.queue_bind` 绑定队列到 Exchange 时，无需指定 Routing Key。



* **Topic Exchange（主题交换机）：**



    Topic Exchange 将消息的 Routing Key 与 Binding Key 进行 **模式匹配**，然后将消息路由到匹配的 Queue。  Binding Key 可以使用通配符：



    * `*` (星号)：匹配一个单词。

    * `#` (井号)：匹配零个或多个单词。



    单词之间通常用 `.` (点号) 分隔。



    **Mermaid 图示：**



    ```mermaid

graph TD
    Producer --> Exchange[Topic Exchange]
    Exchange --> |Routing Key matches kern.*| QueueA[Queue A]
    Exchange --> |Routing Key matches *.critical| QueueB[Queue B]
    QueueA --> Consumer1[Consumer 1]
    QueueB --> Consumer2[Consumer 2]
    style Exchange fill:#f9f,stroke:#333,stroke-width:2px
    style QueueA fill:#ccf,stroke:#333,stroke-width:2px
    style QueueB fill:#ccf,stroke:#333,stroke-width:2px


**代码实践（Python - pika）：**



**生产者 (emit_log_topic.py):**



```python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic') # 声明 Topic Exchange
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
                      routing_key=routing_key, # 使用 routing_key 进行路由
                      body=message.encode())
print(f" [x] Sent {routing_key}:{message}")
connection.close()
```



**消费者 (receive_logs_topic.py):**



```python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic') # 声明 Topic Exchange
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
    sys.exit(1)
for binding_key in binding_keys:
    channel.queue_bind(exchange='topic_logs',
                       queue=queue_name,
                       routing_key=binding_key) # 绑定队列到 Exchange，指定 Binding Key (支持通配符)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
    print(f" [x] {method.routing_key}:{body.decode()}")
channel.basic_consume(queue=queue_name,
                      on_message_callback=callback,
                      auto_ack=True)
channel.start_consuming()
```



**代码详解：**



* **生产者和消费者：** 都声明了 `topic` 类型的 Exchange，名称为 `topic_logs`。

* **生产者：**  `channel.basic_publish` 使用 `routing_key` 参数进行路由，Routing Key 可以是多个单词，例如 `kern.critical`。

* **消费者：**  `channel.queue_bind` 绑定队列到 Exchange 时，可以使用通配符 `*` 和 `#` 指定 Binding Key，例如 `kern.*`、`*.critical`、`kern.#` 等。

Headers Exchange（首部交换机）：

Headers Exchange 不依赖于 Routing Key 进行路由，而是根据消息的 Headers（消息头） 进行路由。在绑定队列到 Headers Exchange 时，可以指定一组键值对作为匹配条件。当消息的 Headers 中包含与 Binding 键值对匹配的项时，消息会被路由到该队列。

Headers Exchange 可以实现更灵活的路由策略，例如根据消息的属性、标签等进行路由。

Mermaid 图示：

graph TD
Producer --> Exchange(Headers Exchange)
Exchange --> |Headers match format: pdf type: report| QueueA(Queue A)
Exchange --> |Headers match type: log level: error| QueueB(Queue B)
QueueA --> Consumer1(Consumer 1)
QueueB --> Consumer2(Consumer 2)
style Exchange fill:#f9f,stroke:#333,stroke-width:2px
style QueueA fill:#ccf,stroke:#333,stroke-width:2px
style QueueB fill:#ccf,stroke:#333,stroke-width:2px





    **代码实践（Python - pika）：**



    **生产者 (emit_log_headers.py):**



    ```python
    import pika
    import sys
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='headers_logs', exchange_type='headers') # 声明 Headers Exchange
    message = ' '.join(sys.argv[2:]) or "Hello World!"
    headers = {}
    if len(sys.argv) > 1:
        headers['format'] = sys.argv[1] # 设置消息头
    channel.basic_publish(exchange='headers_logs',
                          routing_key='', # routing_key 忽略
                          body=message.encode(),
                          properties=pika.BasicProperties(headers=headers)) # 设置消息属性中的 headers
    print(f" [x] Sent {headers}:{message}")
    connection.close()
    ```



    **消费者 (receive_logs_headers.py):**



    ```python
    import pika
    import sys
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='headers_logs', exchange_type='headers') # 声明 Headers Exchange
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    headers_binding = {}
    if len(sys.argv) > 1:
        headers_binding['format'] = sys.argv[1] # 设置 Binding Headers
    channel.queue_bind(exchange='headers_logs',
                           queue=queue_name,
                           arguments=headers_binding) # 绑定队列到 Exchange，指定 Binding Headers
    print(' [*] Waiting for logs. To exit press CTRL+C')
    def callback(ch, method, properties, body):
        print(f" [x] {properties.headers}:{body.decode()}")
    channel.basic_consume(queue=queue_name,
                          on_message_callback=callback,
                          auto_ack=True)
    channel.start_consuming()
    ```



    **代码详解：**



    * **生产者和消费者：** 都声明了 `headers` 类型的 Exchange，名称为 `headers_logs`。

    * **生产者：**  `channel.basic_publish` 的 `properties` 参数中设置了 `headers` 属性，用于传递消息头。

    * **消费者：**  `channel.queue_bind` 绑定队列到 Exchange 时，使用 `arguments` 参数指定 Binding Headers，用于匹配消息头。



**2.3 Queue（队列）：**



Queue 是 RabbitMQ 中用于存储消息的组件。生产者将消息发送到 Exchange，Exchange 根据路由规则将消息路由到一个或多个 Queue 中。消费者从 Queue 中获取消息进行处理。



Queue 具有以下关键特性：



* **消息存储：** Queue 负责存储消息，直到消费者消费消息或消息过期。

* **FIFO（先进先出）：**  默认情况下，Queue 中的消息按照接收顺序进行存储和消费，遵循 FIFO 原则。

* **消息确认（Acknowledgement）：**  消费者可以向 RabbitMQ Broker 发送消息确认，告知 Broker 消息已被成功处理。Broker 收到确认后，才会从 Queue 中删除消息。

* **队列属性：** Queue 可以设置多种属性，例如：

    * **Durable（持久化）：**  持久化队列会在 RabbitMQ Broker 重启后仍然存在。

    * **Exclusive（排他队列）：**  排他队列只能被声明它的连接使用，连接断开后队列自动删除。

    * **Auto-delete（自动删除）：**  当最后一个消费者取消订阅后，队列会自动删除。



**Mermaid 图示：**



```mermaid
    graph TD
        Exchange --> Queue(Queue)
        Queue --> Consumer(Consumer)
        style Queue fill:#ccf,stroke:#333,stroke-width:2px

代码实践（Python - pika）：

声明队列 (producer.py & consumer.py)：

在之前的代码示例中，我们已经多次使用 channel.queue_declare() 声明队列。


# 声明队列，如果队列不存在则创建，如果已存在则直接使用
channel.queue_declare(queue='hello', durable=True) # 声明名为 'hello' 的持久化队列

代码详解：

channel.queue_declare(queue='hello', durable=True) 声明一个名为 hello 的队列。
durable=True 设置队列为持久化队列。

2.4 Binding（绑定）：

Binding 是 Exchange 和 Queue 之间的关联关系。它定义了 Exchange 如何将消息路由到 Queue。 Binding 包含了以下信息：

Exchange 名称： 指定绑定的 Exchange。
Queue 名称： 指定绑定的 Queue。
Binding Key（绑定键）： 用于路由消息的键，其作用取决于 Exchange 类型。例如，Direct Exchange 和 Topic Exchange 需要 Binding Key 进行路由匹配。

Mermaid 图示：

代码实践（Python - pika）：

在之前的 Exchange 代码示例中，我们已经使用了 channel.queue_bind() 创建绑定。

Direct Exchange 绑定：


channel.queue_bind(exchange='direct_logs',
                   queue=queue_name,
                   routing_key=severity) # 使用 routing_key (Binding Key) 进行绑定

Fanout Exchange 绑定：


channel.queue_bind(exchange='logs', queue=queue_name) # 无需 Binding Key

Topic Exchange 绑定：


channel.queue_bind(exchange='topic_logs',
                   queue=queue_name,
                   routing_key=binding_key) # 使用 binding_key (Binding Key, 支持通配符) 进行绑定

Headers Exchange 绑定：


channel.queue_bind(exchange='headers_logs',
                   queue=queue_name,
                   arguments=headers_binding) # 使用 arguments (Binding Headers) 进行绑定

代码详解：

channel.queue_bind() 方法用于创建绑定，需要指定 Exchange 名称、Queue 名称，以及 Binding Key 或 Binding Headers (取决于 Exchange 类型)。

2.5 Connection（连接） & Channel（信道）：

Connection（连接）： Connection 是生产者/消费者客户端与 RabbitMQ Broker 建立的 TCP 连接。一个 Connection 可以包含多个 Channel。
Channel（信道）： Channel 是在 Connection 内部建立的虚拟连接。大部分 RabbitMQ 操作，例如声明 Exchange、声明 Queue、发布消息、订阅消息等，都是在 Channel 上进行的。

为什么要使用 Channel？

建立和维护 TCP 连接的成本较高。如果每个客户端操作都创建一个新的 TCP 连接，会造成资源浪费，并且效率低下。Channel 的出现是为了解决这个问题。一个 Connection 可以复用多个 Channel，每个 Channel 独立工作，共享同一个 TCP 连接。这样可以有效地降低连接开销，提高系统性能。

Mermaid 图示：

代码实践（Python - pika）：

在之前的代码示例中，我们都使用了以下代码创建 Connection 和 Channel：


import pika
# 创建 Connection
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
# 创建 Channel
channel = connection.channel()

代码详解：

pika.BlockingConnection(pika.ConnectionParameters('localhost')) 创建一个阻塞式的 Connection，连接到本地 RabbitMQ Broker。
connection.channel() 创建一个 Channel 对象。

2.6 Consumer（消费者） & Producer（生产者）：

Producer（生产者）： 生产者是负责创建和发送消息的应用程序。生产者将消息发送到 Exchange。
Consumer（消费者）： 消费者是负责接收和处理消息的应用程序。消费者订阅 Queue，并从 Queue 中获取消息进行处理。

Mermaid 图示：

代码实践（Python - pika）：

在之前的代码示例中，我们已经展示了 Producer 和 Consumer 的基本代码结构。

Producer (producer.py):


import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# ... 声明 Exchange 和 Queue (可选) ...
message_body = "Hello, RabbitMQ!"
channel.basic_publish(exchange='direct_logs',
                      routing_key='info',
                      body=message_body)
print(f" [x] Sent '{message_body}'")
connection.close()

Consumer (consumer.py):


import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# ... 声明 Exchange 和 Queue (可选) ...
def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
channel.basic_consume(queue='hello',
                      on_message_callback=callback,
                      auto_ack=True)
channel.start_consuming()

代码详解：

Producer： 负责创建 Connection、Channel，声明 Exchange (可选) 和 Queue (可选)，构建消息，并使用 channel.basic_publish() 发送消息。
Consumer： 负责创建 Connection、Channel，声明 Exchange (可选) 和 Queue (可选)，定义消息处理回调函数 callback，并使用 channel.basic_consume() 订阅队列，开始消费消息。

总结

本文深入解析了 RabbitMQ 的核心概念，包括 Message、Exchange、Queue、Binding、Connection、Channel、Consumer 和 Producer。通过代码实践和 Mermaid 图表，我们希望您对 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

整理：灏天文库整理

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