Python 简单使用 RabbitMQ

一、安装

pip install pika

二、推送消息到队列中

执行pythone方法

import pika
import time# 用户名和密码
user_info = pika.PlainCredentials('admin','admin')# 连接服务器上的rabbitMQ服务
connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1', 5672, '/', user_info))
# connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))# 创建一个channel
channel = connection.channel()# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，官方推荐，每次使用时都可以加上这句
channel.queue_declare(queue='pythone.test')# 推送消息到队列  
# exchange:当前是一个简单模式，所以这里设置为空字符串就可以了。   
# routing_key：指定消息要发送到哪个queue。 
# body：指定要发送的消息。
channel.basic_publish(exchange='',routing_key='pythone.test',body='{}'.format('test xxx'))# 关闭连接
connection.close()

查看rabbitMQ网页后台

执行后我们进入rabbitMQ网页端后台查看pythone.test 队列已经被创建

并且我们执行了三次，此处产生3条数据未被消费，还被压在队列中。

查看队列内消息列表

我们改造一下，将推送消息放到方法中。

三、封装成生产者、消费者方法

生产者product：

import pika
import time# 用户名和密码
user_info = pika.PlainCredentials('admin','admin')# 连接服务器上的rabbitMQ服务
connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1', 5672, '/', user_info))
# connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))# 创建一个channel
channel = connection.channel()# 生产者方法
def product():print("进入生产者方法！")# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，官方推荐，每次使用时都可以加上这句channel.queue_declare(queue='pythone.test')# 推送消息到队列  # exchange:当前是一个简单模式，所以这里设置为空字符串就可以了。   # routing_key：指定消息要发送到哪个queue。 # body：指定要发送的消息。channel.basic_publish(exchange='',routing_key='pythone.test',body='{}'.format('test xxx'))# 关闭连接connection.close()if __name__ == '__main__':start_time = time.time()    # 程序开始时间print("========start=========|"+str(start_time))product()end_time = time.time()    # 程序结束时间print("========end===========|"+str(end_time))

消费者consumer：

# 消费者方法
def consumer():print("进入消费者方法！")# 消费队列内的消息# queue：接收指定queue的消息# auto_ack：指定为True，表示消息接收到后自动给消息发送方回复确认，已收到消息# on_message_callback：设置收到消息的回调函数channel.basic_consume(queue='pythone.test', auto_ack=True, on_message_callback=mq_consumer_callback)# 一直处于等待接收消息的状态，如果没收到消息就一直处于阻塞状态，收到消息就调用上面的回调函数channel.start_consuming()# 消费者收到消息调用的回调函数
# channel: 包含channel的一切属性和方法
# method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
# properties: basic_publish 通过 properties 传入的参数
# body: basic_publish发送的消息
def mq_consumer_callback(ch, method, properties, body):print('消费者收到:{}'.format(body))

TestRabbitMQ.py

import pika
import time# 用户名和密码
user_info = pika.PlainCredentials('admin','admin')# 连接服务器上的rabbitMQ服务
connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1', 5672, '/', user_info))
# connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))# 创建一个channel
channel = connection.channel()# 生产者方法
def product():print("进入生产者方法！")# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，官方推荐，每次使用时都可以加上这句channel.queue_declare(queue='pythone.test')# 推送消息到队列  # exchange:当前是一个简单模式，所以这里设置为空字符串就可以了。   # routing_key：指定消息要发送到哪个queue。 # body：指定要发送的消息。channel.basic_publish(exchange='',routing_key='pythone.test',body='{}'.format('test xxx'))# 关闭连接connection.close()# 消费者方法
def consumer():print("进入消费者方法！")# 消费队列内的消息# queue：接收指定queue的消息# auto_ack：指定为True，表示消息接收到后自动给消息发送方回复确认，已收到消息# on_message_callback：设置收到消息的回调函数channel.basic_consume(queue='pythone.test', auto_ack=True, on_message_callback=mq_consumer_callback)# 一直处于等待接收消息的状态，如果没收到消息就一直处于阻塞状态，收到消息就调用上面的回调函数channel.start_consuming()# 消费者收到消息调用的回调函数
# channel: 包含channel的一切属性和方法
# method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
# properties: basic_publish 通过 properties 传入的参数
# body: basic_publish发送的消息
def mq_consumer_callback(ch, method, properties, body):print('消费者收到:{}'.format(body))if __name__ == '__main__':start_time = time.time()    # 程序开始时间print("========start=========|"+str(start_time))# product()consumer()end_time = time.time()    # 程序结束时间print("========end===========|"+str(end_time))