Python pika 模块：进阶教程

python库 › 消息队列
, ,

Python pika 模块进阶教程

Pika 是一个 Python 客户端库，用于与 RabbitMQ 通信。RabbitMQ 是一种开源的消息代理软件，支持多种消息协议。Pika 专为使用 RabbitMQ 提供了强大的功能，能够轻松地实现消息的发送和接收。它兼容 Python 3.x，并承载着众多现代应用的消息传递需求。Pika 模块的设计旨在简化与 RabbitMQ 的交互，同时具备高度的灵活性和可扩展性，适用于从简单的应用到复杂的分布式系统。

模块介绍

Pika 是纯 Python 实现的 RabbitMQ 客户端，支持多种传输协议和消息传递模式。它提供了异步通信的能力，适合需要高效处理大量消息的应用场景。适配的 Python 版本为 Python 3.6 及以上。Pika 模块不仅支持基本的 RPC 和发布 - 订阅模式，还支持高级功能如确认交付和回调机制，使得开发者能够创建可扩展且可靠的消息传递系统。

应用场景

Pika 模块主要用于需要异步消息传递的应用场景。其广泛应用于以下几个方面：

  1. 高并发服务：在高并发的 Web 服务中，Pika 可以支持通过消息队列处理请求，避免直接操作数据库带来的瓶颈。
  2. 微服务架构：在微服务架构中，服务间需要高效的通信，Pika 使服务间的信息交互得以轻松实现。
  3. 任务队列：在后台处理任务（如数据处理、图像处理等）时，Pika 可用于管理任务队列，确保任务的顺序执行及结果的及时返回。

安装说明

Pika 模块并不是 Python 的默认模块，需通过 pip 进行安装。你可以在终端中使用以下命令进行安装：

1
pip install pika  # 安装Pika模块

用法举例

1. 基本的发送和接收消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import pika  # 导入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()

这段代码展示了如何发送一个简单的消息到名为 “hello” 的队列。通过连接到 RabbitMQ 服务器，声明队列，并调用 basic_publish 方法发送消息。

2. 接收消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import pika  # 导入pika库

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

# 声明同样的队列用于接收
channel.queue_declare(queue='hello')

# 定义回调函数，用于处理接收到的消息
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)  # 打印接收到的消息

# 设置队列消费消息
channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')  # 提示正在等待消息
channel.start_consuming()  # 开始消费

该示例展示了接收 “hello” 队列中的消息，并通过回调函数处理它们。basic_consume 方法可用于设置消息的消费模式。

3. 使用 RPC 模式实现远程调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import pika  # 导入pika库
import uuid  # 导入uuid模块

# 创建一个RPC客户端类
class RpcClient:
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()  # 创建频道
        
        # 声明RPC队列
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue  # 获取回调队列名
        
        # 定义响应的回调函数
        self.channel.basic_consume(queue=self.callback_queue, 
                                   on_message_callback=self.on_response, 
                                   auto_ack=True)

    def on_response(self, ch, method, properties, body):
        self.response = body  # 保存响应内容

    def call(self, n):
        self.response = None
        self.channel.basic_publish(exchange='',
                                    routing_key='rpc_queue',
                                    properties=pika.BasicProperties(reply_to=self.callback_queue),
                                    body=str(n))  # 发送rpc请求
        while self.response is None:  # 等待结果
            self.connection.process_data_events()  # 处理消息
        return int(self.response)  # 返回结果

rpc_client = RpcClient()  # 创建RPC客户端实例
print(" [x] Requesting fib(30)")  # 控制台提示请求
response = rpc_client.call(30)  # 调用RPC方法
print(" [.] Got %r" % response)  # 打印结果

这个示例展示了如何使用 Pika 模块完成 RPC（远程过程调用）。它定义了一个 RpcClient 类，发送请求，并等待响应的过程。

强烈建议大家关注本人的博客全糖冲击博客，我会定期更新 Python 标准库使用教程，方便大家查询和学习。我的博客不仅涉及基础知识，还有深入的实用案例分析，可以帮助你更好地理解和运用 Python 编程。希望你能在这里找到想要的内容，提升自己的编程能力，成为更优秀的开发者！