利用Python操作消息队列RabbitMQ的方法教程

前言

RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统。他遵循Mozilla Public License开源协议。
MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信,而不是通过直接调用彼此来通信,直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

应用场景:

RabbitMQ无疑是目前最流行的消息队列之一,对各种语言环境的支持也很丰富,作为一个.NET developer有必要学习和了解这一工具。消息队列的使用场景大概有3种:

1、系统集成,分布式系统的设计。各种子系统通过消息来对接,这种解决方案也逐步发展成一种架构风格,即“通过消息传递的架构”。

2、当系统中的同步处理方式严重影响了吞吐量,比如日志记录。假如需要记录系统中所有的用户行为日志,如果通过同步的方式记录日志势必会影响系统的响应速度,当我们将日志消息发送到消息队列,记录日志的子系统就会通过异步的方式去消费日志消息。

3、系统的高可用性,比如电商的秒杀场景。当某一时刻应用服务器或数据库服务器收到大量请求,将会出现系统宕机。如果能够将请求转发到消息队列,再由服务器去消费这些消息将会使得请求变得平稳,提高系统的可用性。

一、安装环境

首先是在 Linux 上安装 rabbitmq

# 环境为CentOS 7
yum install rabbitmq-server # 安装RabbitMQ
systemctl start rabbitmq-server # 启动
systemctl enable rabbitmq-server # 开机自启
systemctl stop firewall-cmd  # 临时关闭防火墙

然后用 pip 安装 Python3 的开发包

pip3 install pika

安装好软件之后可以访问http://115.xx.xx.xx:15672/来访问自带的 web 页面来查看和管理 RabbitMQ。默认管理员的用户密码都是guest

二、简单的向队列中加入消息

#!/usr/bin/env python3
# coding=utf-8
# @Time : 2017/6/13 19:25
# @Author : Shawn
# @Blog : https://blog.just666.cn
# @Email : shawnbluce@gmail.com
# @purpose : RabbitMQ_Producer
import pika
# 创建连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters(host='115.xx.xx.xx'))
# 创建频道对象
channel = connection.channel()
# 指定一个队列,如果该队列不存在则创建
channel.queue_declare(queue='test_queue')
# 提交消息
for i in range(10):
 channel.basic_publish(exchange='', routing_key='test_queue', body='hello,world' + str(i))
 print("sent...")
# 关闭连接
connection.close()

三、简单的从队列中获取消息

#!/usr/bin/env python3
# coding=utf-8
# @Time : 2017/6/13 19:40
# @Author : Shawn
# @Blog : https://blog.just666.cn
# @Email : shawnbluce@gmail.com
# @purpose : RabbitMQ_Consumer
import pika
credentials = pika.PlainCredentials('guest', 'guest')
# 连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('115.xx.xx.xx', 5672, '/', credentials))
channel = connection.channel()
# 指定一个队列,如果该队列不存在则创建
channel.queue_declare(queue='test_queue')
# 定义一个回调函数
def callback(ch, method, properties, body):
 print(body.decode('utf-8'))
# 告诉RabbitMQ使用callback来接收信息
channel.basic_consume(callback, queue='test_queue', no_ack=False)
print('waiting...')
# 开始接收信息,并进入阻塞状态,队列里有信息才会调用callback进行处理。按ctrl+c退出。
channel.start_consuming()

四、万一消费者掉线了

想象这样一种情况:

消费者从消息队列中获取了 n 条数据,正要处理呢结果宕机了,那该怎么办?在 RabbieMQ 中有一个 ACK 可以用来确认消费者处理结束。就有点类似网络中的 ACK,消费者每次从队列中获取了数据之后队列不会立刻将数据移除,而是等待对应的 ACK。消费者获取到数据并处理完成之后会向队列发送一个 ACK 包,通知 RabbitMQ 这堆消息已经处理妥当了,可以删除了,这时候 RabbitMQ 才会将数据从队列中移除。所以这种情况下即使消费者掉线也没有什么问题,数据依旧会在队列中存在,留给其他消费者处理。

在 Python 中这样实现:

消费者有这样一行代码channel.basic_consume(callback, queue='test_queue', no_ack=False) ,其中no_ack=False表示不发送确认包。将其修改为no_ack=True就会在每次处理完之后向 RabbitMQ 发送一个确认包,以确认消息处理完毕。

五、万一 RabbitMQ 宕机了呢

虽然有了 ACK 包,但是万一 RabbitMQ 挂了那数据还是会损失。所以我们可以给 RabbitMQ 设置一个数据持久化存储。RabbitMQ 会将数据持久化存储到磁盘上,保证下次再启动的时候队列还在。

在 Python 中这样实现:

我们声明一个队列是这样的channel.queue_declare(queue='test_queue') ,如果需要持久化一个队列可以这样声明channel.queue_declare(queue='test_queue', durable=True) 。不过这行直接放在代码中是不能执行的,因为以前已经有了一个名为test_queue的队列,RabbitMQ 不允许用不同的方式声明同一个队列,所以可以换一个队列名新建来指定数据持久化存储。不过如果只是这样声明的话,在 RabbitMQ 宕机重启后确实队列还在,不过队列里的数据就没有了。除非我们这样来声明队列channel.basic_publish(exchange='', routing_key="test_queue", body=message, properties=pika.BasicProperties(delivery_mode = 2,))

六、最简单的发布订阅

最简单的发布订阅在 RabbitMQ 中称之为Fanout模式。也就是说订阅者订阅某个频道,然后发布者向这个频道中发布消息,所有订阅者就都能接收到这条消息。不过因为发布者需要使用订阅者创建的随机队列所以需要先启动订阅者才能启动发布者。

发布者代码:

#!/usr/bin/env python3
# coding=utf-8
# @Time : 2017/6/13 20:21
# @Author : Shawn
# @Blog : https://blog.just666.cn
# @Email : shawnbluce@gmail.com
# @purpose : RabbitMQ_Publisher
import pika
# 创建连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters(host='115.xx.xx.xx'))
# 创建频道对象
channel = connection.channel()
# 定义交换机,exchange表示交换机名称,type表示类型
channel.exchange_declare(exchange='my_fanout',
       type='fanout')
message = 'Hello Python'
# 将消息发送到交换机
channel.basic_publish(exchange='my_fanout', # 指定exchange
      routing_key='', # fanout下不需要配置,配置了也不会生效
      body=message)
connection.close()

订阅者代码:

#!/usr/bin/env python3
# coding=utf-8
# @Time : 2017/6/13 20:20
# @Author : Shawn
# @Blog : https://blog.just666.cn
# @Email : shawnbluce@gmail.com
# @purpose : RabbitMQ_Subscriber
import pika
credentials = pika.PlainCredentials('guest', 'guest')
# 连接到RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('115.xx.xx.xx', 5672, '/', credentials))
channel = connection.channel()
# 定义交换机,进行exchange声明,exchange表示交换机名称,type表示类型
channel.exchange_declare(exchange='my_fanout',
       type='fanout')
# 随机创建队列
result = channel.queue_declare(exclusive=True) # exclusive=True表示建立临时队列,当consumer关闭后,该队列就会被删除
queue_name = result.method.queue
# 将队列与exchange进行绑定
channel.queue_bind(exchange='my_fanout',
     queue=queue_name)
# 定义回调方法
def callback(ch, method, properties, body):
 print(body.decode('utf-8'))
# 从队列获取信息
channel.basic_consume(callback,
      queue=queue_name,
      no_ack=True)
channel.start_consuming()

总结

以上就是这篇文章的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,如果有疑问大家可以留言交流,谢谢大家对我们的支持。

时间: 2017-07-18

python使用rabbitmq实现网络爬虫示例

编写tasks.py 复制代码 代码如下: from celery import Celeryfrom tornado.httpclient import HTTPClientapp = Celery('tasks')app.config_from_object('celeryconfig')@app.taskdef get_html(url):    http_client = HTTPClient()    try:        response = http_client.fetch(u

Python操作RabbitMQ服务器实现消息队列的路由功能

Python使用Pika库(安装:sudo pip install pika)可以操作RabbitMQ消息队列服务器(安装:sudo apt-get install rabbitmq-server),这里我们来看一下MQ相关的路由功能. 路由键的实现 比如有一个需要给所有接收端发送消息的场景,但是如果需要自由定制,有的消息发给其中一些接收端,有些消息发送给另外一些接收端,要怎么办呢?这种情况下就要用到路由键了. 路由键的工作原理:每个接收端的消息队列在绑定交换机的时候,可以设定相应的路由键.发送

详解Python操作RabbitMQ服务器消息队列的远程结果返回

先说一下笔者这里的测试环境:Ubuntu14.04 + Python 2.7.4 RabbitMQ服务器 sudo apt-get install rabbitmq-server Python使用RabbitMQ需要Pika库 sudo pip install pika 远程结果返回 消息发送端发送消息出去后没有结果返回.如果只是单纯发送消息,当然没有问题了,但是在实际中,常常会需要接收端将收到的消息进行处理之后,返回给发送端. 处理方法描述:发送端在发送信息前,产生一个接收消息的临时队列,该队

利用Python学习RabbitMQ消息队列

RabbitMQ可以当做一个消息代理,它的核心原理非常简单:即接收和发送消息,可以把它想象成一个邮局:我们把信件放入邮箱,邮递员就会把信件投递到你的收件人处,RabbitMQ就是一个邮箱.邮局.投递员功能综合体,整个过程就是:邮箱接收信件,邮局转发信件,投递员投递信件到达收件人处. RabbitMQ和邮局的主要区别就是RabbitMQ接收.存储和发送的是二进制数据----消息. rabbitmq基本管理命令: 一步启动Erlang node和Rabbit应用:sudo rabbitmq-serv

Python+Pika+RabbitMQ环境部署及实现工作队列的实例教程

rabbitmq中文翻译的话,主要还是mq字母上:Message Queue,即消息队列的意思.前面还有个rabbit单词,就是兔子的意思,和python语言叫python一样,老外还是蛮幽默的.rabbitmq服务类似于mysql.apache服务,只是提供的功能不一样.rabbimq是用来提供发送消息的服务,可以用在不同的应用程序之间进行通信. 安装rabbitmq 先来安装下rabbitmq,在ubuntu 12.04下可以直接通过apt-get安装: sudo apt-get insta

Python通过RabbitMQ服务器实现交换机功能的实例教程

快速回顾一下RabbitMQ服务器的安装: sudo apt-get install rabbitmq-server Python使用RabbitMQ需要Pika库: sudo pip install pika 好了,接下来我们先看交换机的工作原理:消息发送端先将消息发送给交换机,交换机再将消息发送到绑定的消息队列,而后每个接收端都能从各自的消息队列里接收到信息. 下面用send.py和receive.py来模拟实现交换机的功能.send.py表示发送端,receive.py表示接收端. rec

Python实现微信消息防撤回功能的实例代码

微信(WeChat)是腾讯公司于2011年1月21日推出的一款社交软件,8年时间微信做到日活10亿,日消息量450亿.在此期间微信也推出了不少的功能如:"摇一摇"."漂流瓶"."朋友圈"."附近的人"."公众平台"."小程序"等等,涵盖了我们生活的方方面面,微信正在慢慢践行着他们的口号:微信,是一个生活方式 一.背景介绍 产品的更新迭代必然会伴随着功能的推出和下线,今天我们要讲的便是微信

Python模拟百度自动输入搜索功能的实例

如下所示: # 访问百度,模拟自动输入搜索 # 代码中引入selenium版本为:3.4.3 # 通过Chrom浏览器访问发起请求 # Chrom版本:59 ,chromdriver:2.3 # 需要对应版本的Chrom和chromdriver # 请联系QQ:878799579 from selenium import webdriver # 引入Keys类包 发起键盘操作 from selenium.webdriver.common.keys import Keys import time

基于python tkinter的点名小程序功能的实例代码

代码如下所示: import datetime import json import os import random import tkinter as tk import openpyxl # 花名册文件名 excel_file_path = "花名册.xlsx"#需在当前目录创建对应花名册.xlsx # 工作表名 excel_sheet = "Sheet1" # 记录存储文件名 file_path = "name_record.json"

Python实现基于多线程、多用户的FTP服务器与客户端功能完整实例

本文实例讲述了Python实现基于多线程.多用户的FTP服务器与客户端功能.分享给大家供大家参考,具体如下: 项目介绍: 1. 用户加密认证 2. 允许同时多用户登录 3. 每个用户有自己的家目录 ,且只能访问自己的家目录 4. 对用户进行磁盘配额,每个用户的可用空间不同 5. 允许用户在ftp server上随意切换目录 6. 允许用户查看当前目录下文件 7. 允许上传和下载文件,保证文件一致性 8. 文件传输过程中显示进度条 实现的原理: 服务器端启用端口监听,并对每一连接启用一个线程,对用

Python多进程与服务器并发原理及用法实例分析

本文实例分析了Python多进程与服务器并发原理及用法.分享给大家供大家参考,具体如下: 进程 什么是进程 进程:正在进行的一个过程或者说一个任务.而负责执行任务则是cpu. 进程与程序的区别 程序仅仅只是一堆代码而已,而进程指的是程序的运行过程. 并发与并行 无论是并行还是并发,在用户看来都是'同时'运行的,不管是进程还是线程,都只是一个任务而已,真是干活的是cpu,cpu来做这些任务,而一个cpu同一时刻只能执行一个任务 一 并发:是伪并行,即看起来是同时运行.单个cpu+多道技术就可以实现

Python selenium文件上传下载功能代码实例

上传 html文件内容如下:操作步骤 <html> <head> <meta http-equiv="content-type" content="text/html;charset=utf-8" /> <title>upload_file</title> <script type="text/javascript" async="" src="htt

基于Python实现体育彩票选号器功能代码实例

一,概要 需求: 实现一个GUI界面下的 6+1体育彩票选号器. (1) 要求界面可以加载系统时间及开奖时间 (2) 功能区完成人选及机选的功能 人选 --> 手动输入6+1位数字.前6位必须在0-9之间的数字.后1位必须是0-4之间的数字 机选 --> (1) 填写数量(加校验必须为数字且不能为空)点击开始按钮后把选举的数字添加到展示界面中 (2) 允许选举的数字重复及不重复两种选择 (3) 展示区的设置,显示已选的彩票号码 (4) 完成清空展示区内容功能 (5) 完成关闭整个界面窗口功能