RabbitMQ快速上手_rabbitmq python

现在市场上主流的MQ有很多，比如 ActiveMQ、RabbitMQ、RocketMQ、Kafka、ZeroMQ 等。

RabbitMQ 是一个流行的开源消息代理，它实现了高级消息队列协议（AMQP）。它被广泛应用于分布式系统中，能够解耦应用程序、分布式任务和负载平衡。其有着非常丰富的特性和优势：高可靠性、路由灵活、集群扩张性高、高可用、支持多种协议、支持多种客户端和有着丰富的插件系统。

RabbitMQ目前被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。

RabbitMQ 是一个流行的开源消息代理，它实现了高级消息队列协议（AMQP）。它被广泛应用于分布式系统中，能够解耦应用程序、分布式任务和负载平衡。下面是快速上手 RabbitMQ 的指南，包括安装、基本概念和基本操作。

1. 安装 RabbitMQ

在 Windows 上安装

1. 下载和安装 Erlang：
   • RabbitMQ 依赖于 Erlang，所以需要先安装 Erlang。
   • 按照安装向导完成安装。
2. 下载和安装 RabbitMQ：
   • 下载地址：RabbitMQ 下载
   • 下载完成后，运行安装程序并按照提示完成安装。

在 Ubuntu 上安装

# 更新软件包列表
sudo apt-get update

# 安装 Erlang
sudo apt-get install -y erlang

# 安装 RabbitMQ
sudo apt-get install -y rabbitmq-server

# 启动 RabbitMQ 服务
sudo systemctl start rabbitmq-server

# 设置 RabbitMQ 服务开机自启动
sudo systemctl enable rabbitmq-server
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2. 基本概念

• Producer（生产者）：发送消息的程序。
• Consumer（消费者）：接收消息的程序。
• Queue（队列）：存储消息的缓冲区。
• Exchange（交换器）：接收消息并将它们路由到队列。
• Binding（绑定）：交换器和队列之间的链接。
• Routing Key（路由键）：用于交换器将消息路由到特定队列的键。

3. 管理 RabbitMQ

RabbitMQ 提供了一个基于 Web 的管理插件，可以方便地管理和监控 RabbitMQ 服务器。

# 启用管理插件
sudo rabbitmq-plugins enable rabbitmq_management

# 访问管理界面
# 在浏览器中访问：http://localhost:15672/
# 默认用户名和密码都是 guest
1
2
3
4
5
6

4.使用场景

1. 异步处理

使用场景：当一个任务需要较长时间处理时，将任务放入队列，异步处理，避免阻塞主线程。

示例：订单处理系统中，用户提交订单后，将订单信息放入队列，后台异步处理订单。

Client -> [Order Service] -> [RabbitMQ Queue] -> [Order Processing Service]
1

2. 任务队列

使用场景：将任务分发给多个消费者，实现负载均衡和并行处理。

示例：图像处理系统中，用户上传图片后，将图片处理任务放入队列，多个消费者同时处理多个图片任务。

Client -> [Image Upload Service] -> [RabbitMQ Queue] -> [Image Processing Workers]
1

3. 日志聚合

使用场景：收集分布式系统中的日志，统一存储和分析。

示例：多个微服务将日志发送到 RabbitMQ，日志聚合服务从队列中读取日志并存储到数据库或日志管理系统。

[Microservice A] -> 
[Microservice B] -> [RabbitMQ Queue] -> [Log Aggregation Service]
[Microservice C] ->
1
2
3

4. 消息广播

使用场景：将消息广播给多个消费者，实现消息的多播。

示例：实时通知系统中，服务器将通知消息广播给多个客户端。

[Notification Service] -> [RabbitMQ Exchange] -> [Client A]
                                          -> [Client B]
                                          -> [Client C]
1
2
3

5. 微服务通信

使用场景：在微服务架构中，使用 RabbitMQ 作为服务间通信的中介，解耦服务。

示例：电商系统中，订单服务和库存服务通过 RabbitMQ 进行通信。

[Order Service] -> [RabbitMQ Exchange] -> [Inventory Service]
1

6. 数据流处理

使用场景：处理实时数据流，将数据流分发给多个处理节点。

示例：实时数据分析系统中，数据流从数据源进入 RabbitMQ，多个数据处理节点从队列中读取数据进行分析。

[Data Source] -> [RabbitMQ Queue] -> [Data Processing Node A]
                                  -> [Data Processing Node B]
1
2

7. 延迟任务

使用场景：在指定时间后执行任务。

示例：发送延迟通知或邮件，任务放入 RabbitMQ 的延迟队列，延迟指定时间后再处理。

[Task Scheduler] -> [RabbitMQ Delay Queue] -> [Notification Service]
1

8. 事件驱动架构

使用场景：基于事件的系统架构设计，通过消息队列传递事件。

示例：用户注册系统，用户注册成功后，触发多个后续处理，如发送欢迎邮件、更新统计数据等。

[User Registration Service] -> [RabbitMQ Exchange] -> [Email Service]
                                                  -> [Analytics Service]
                                                  -> [Profile Service]
1
2
3

9. 分布式事务

使用场景：通过消息队列实现分布式系统中的事务一致性。

示例：订单服务和支付服务之间，通过 RabbitMQ 确保订单和支付的一致性。

[Order Service] -> [RabbitMQ Queue] -> [Payment Service]
1

5.主要特点

可靠性（Reliability）：RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认。

灵活的路由（Flexible Routing）：在消息进入队列之前，通过 Exchange 来路由消息的。对于典型的路由功能，RabbitMQ 已经提供了一些内置的
Exchange 来实现。针对更复杂的路由功能，可以将多个 Exchange 绑定在一起，也通过插件机制实现自己的 Exchange 。

消息集群（Clustering）：多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker 。

高可用（Highly Available Queues）：队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。

多种协议（Multi-protocol）：RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等等。

多语言客户端（Many Clients）：RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等。

管理界面（Management UI）:RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。

跟踪机制（Tracing）:如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。

插件机制（Plugin System）:RabbitMQ 提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。

6. 使用示例

下面是使用 Python 和 pika 库的一个简单示例，展示如何发送和接收消息。

安装 pika

pip install pika
1

生产者（发送消息）

pythonimport pika

# 连接到 RabbitMQ 服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建一个队列
channel.queue_declare(queue='hello')

# 发送消息到队列
channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
print(" [x] Sent 'Hello World!'")

# 关闭连接
connection.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

消费者（接收消息）

pythonimport pika

# 连接到 RabbitMQ 服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建一个队列
channel.queue_declare(queue='hello')

# 定义回调函数，处理从队列接收到的消息
def callback(ch, method, properties, body):
    print(f" [x] Received {body}")

# 告诉 RabbitMQ 用回调函数来接收消息
channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

5. 常用命令

• 查看队列：

  sudo rabbitmqctl list_queues
  1

• 添加用户：

  sudo rabbitmqctl add_user myuser mypassword
  1

• 设置用户权限：

  sudo rabbitmqctl set_permissions -p / myuser ".*" ".*" ".*"
  1

• 删除用户：

  sudo rabbitmqctl delete_user myuser
  1