Python开发【十一章】：RabbitMQ队列
RabbitMQ队列
rabbitMQ是消息队列；想想之前的我们学过队列queue：threading queue（线程queue，多个线程之间进行数据交互）、进程queue（父进程与子进程进行交互或者同属于同一父进程下的多个子进程进行交互）；如果两个独立的程序，那么之间是不能通过queue进行交互的,这时候我们就需要一个中间代理即rabbitMQ
消息队列：
- RabbitMQ
- ZeroMQ
- ActiveMQ
- ...........
原理：
1、安装和基本使用
安装RabbitMQ服务 http://www.rabbitmq.com/install-standalone-mac.html
python安装RabbitMQ模块
pip install pika
or
easy_install pika
or
源码https://pypi.python.org/pypi/pika2、实现最简单的队列通信
发送端：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel() #声明一个管道(管道内发消息)channel.queue_declare(queue='lzl') #声明queue队列channel.basic_publish(exchange='',routing_key='lzl', #routing_key 就是queue名body='Hello World!'
)
print("Sent 'Hello,World!'")
connection.close() #关闭
接收端：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel()channel.queue_declare(queue='lzl')def callback(ch,method,properties,body):print(ch,method,properties)#ch:<pika.adapters.blocking_connection.BlockingChannel object at 0x002E6C90> 管道内存对象地址#methon:<Basic.Deliver(['consumer_tag=ctag1.03d155a851b146f19cee393ff1a7ae38', #具体信息# 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=lzl'])>#properties:<BasicProperties>print("Received %r"%body)channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息queue="lzl",no_ack=True) #接受到消息后不返回ack，无论本地是否处理完消息都会在队列中消失
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #开始收消息
注：windows连linux上的rabbitMQ会出现报错，需要提供用户名密码
3、RabbitMQ消息分发轮询
先启动消息生产者，然后再分别启动3个消费者，通过生产者多发送几条消息，你会发现，这几条消息会被依次分配到各个消费者身上
在这种模式下，RabbitMQ会默认把p发的消息公平的依次分发给各个消费者(c),跟负载均衡差不多
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel() #声明一个管道(管道内发消息)
channel.queue_declare(queue='lzl') #声明queue队列
channel.basic_publish(exchange='',routing_key='lzl', #routing_key 就是queue名body='Hello World!'
)
print("Sent 'Hello,World!'")
connection.close() #关闭pubulish.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel()channel.queue_declare(queue='lzl')def callback(ch,method,properties,body):print(ch,method,properties)#ch:<pika.adapters.blocking_connection.BlockingChannel object at 0x002E6C90> 管道内存对象地址#methon:<Basic.Deliver(['consumer_tag=ctag1.03d155a851b146f19cee393ff1a7ae38', #具体信息# 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=lzl'])>#properties:<BasicProperties>print("Received %r"%body)channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息queue="lzl",no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #开始收消息consume.py
通过执行pubulish.py和consume.py可以实现上面的消息公平分发，那假如c1收到消息之后宕机了，会出现什么情况呢？rabbitMQ是如何处理的？现在我们模拟一下
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel() #声明一个管道(管道内发消息)
channel.queue_declare(queue='lzl') #声明queue队列
channel.basic_publish(exchange='',routing_key='lzl', #routing_key 就是queue名body='Hello World!'
)
print("Sent 'Hello,World!'")
connection.close() #关闭publish.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pika,timeconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel()channel.queue_declare(queue='lzl')def callback(ch,method,properties,body):print("->>",ch,method,properties)time.sleep(15) # 模拟处理时间print("Received %r"%body)channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息queue="lzl",no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #开始收消息consume.py
在consume.py的callback函数里增加了time.sleep模拟函数处理，通过上面程序进行模拟发现，c1接收到消息后没有处理完突然宕机，消息就从队列上消失了，rabbitMQ把消息删除掉了；如果程序要求消息必须要处理完才能从队列里删除，那我们就需要对程序进行处理一下
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel() #声明一个管道(管道内发消息)
channel.queue_declare(queue='lzl') #声明queue队列
channel.basic_publish(exchange='',routing_key='lzl', #routing_key 就是queue名body='Hello World!'
)
print("Sent 'Hello,World!'")
connection.close() #关闭publish.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pika,timeconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel()channel.queue_declare(queue='lzl')def callback(ch,method,properties,body):print("->>",ch,method,properties)#time.sleep(15) # 模拟处理时间print("Received %r"%body)ch.basic_ack(delivery_tag=method.delivery_tag)channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息queue="lzl",)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #开始收消息consume.py
通过把consume.py接收端里的no_ack=True去掉之后并在callback函数里面添加ch.basic_ack(delivery_tag = method.delivery_tag，就可以实现消息不被处理完不能在队列里清除
查看消息队列数：
4、消息持久化
如果消息在传输过程中rabbitMQ服务器宕机了，会发现之前的消息队列就不存在了，这时我们就要用到消息持久化，消息持久化会让队列不随着服务器宕机而消失，会永久的保存下去
发送端：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel() #声明一个管道(管道内发消息)channel.queue_declare(queue='lzl',durable=True) #队列持久化channel.basic_publish(exchange='',routing_key='lzl', #routing_key 就是queue名body='Hello World!',properties=pika.BasicProperties(delivery_mode = 2 #消息持久化)
)
print("Sent 'Hello,World!'")
connection.close() #关闭接收端：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pika,timeconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel()channel.queue_declare(queue='lzl',durable=True)def callback(ch,method,properties,body):print("->>",ch,method,properties)time.sleep(15) # 模拟处理时间print("Received %r"%body)ch.basic_ack(delivery_tag=method.delivery_tag)channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息queue="lzl",)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #开始收消息
5、消息公平分发
如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，可以在各个消费者端，配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了
channel.basic_qos(prefetch_count=1)
带消息持久化+公平分发
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel() #声明一个管道(管道内发消息)
channel.queue_declare(queue='lzl',durable=True) #队列持久化
channel.basic_publish(exchange='',routing_key='lzl', #routing_key 就是queue名body='Hello World!',properties=pika.BasicProperties(delivery_mode = 2 #消息持久化
)
)
print("Sent 'Hello,World!'")
connection.close() #关闭pubulish.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pika,timeconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel()channel.queue_declare(queue='lzl',durable=True)def callback(ch,method,properties,body):print("->>",ch,method,properties)time.sleep(15) # 模拟处理时间print("Received %r"%body)ch.basic_ack(delivery_tag=method.delivery_tag)channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息queue="lzl",)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #开始收消息consume.py
6、Publish\Subscribe(消息发布\订阅)
之前的例子都基本都是1对1的消息发送和接收，即消息只能发送到指定的queue里，但有些时候你想让你的消息被所有的Queue收到，类似广播的效果，这时候就要用到exchange了，
An exchange is a very simple thing. On one side it receives messages from producers and the other side it pushes them to queues. The exchange must know exactly what to do with a message it receives. Should it be appended to a particular queue? Should it be appended to many queues? Or should it get discarded. The rules for that are defined by the exchange type.
Exchange在定义的时候是有类型的，以决定到底是哪些Queue符合条件，可以接收消息
fanout: 所有bind到此exchange的queue都可以接收消息
direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息
topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息
headers: 通过headers 来决定把消息发给哪些queue
表达式符号说明：#代表一个或多个字符，*代表任何字符
例：#.a会匹配a.a，aa.a，aaa.a等
*.a会匹配a.a，b.a，c.a等
注：使用RoutingKey为#，Exchange Type为topic的时候相当于使用fanout
① fanout接收所有广播：广播表示当前消息是实时的，如果没有一个消费者在接受消息，消息就会丢弃，在这里消费者的no_ack已经无用，因为fanout不会管你处理消息结束没有，发过的消息不会重发，记住广播是实时的
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='logs',type='fanout')message = "info: Hello World!"
channel.basic_publish(exchange='logs',routing_key='', #广播不用声明queuebody=message)
print(" [x] Sent %r" % message)
connection.close()publish.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='logs',type='fanout')result = channel.queue_declare(exclusive=True) # 不指定queue名字,rabbit会随机分配一个名字,# exclusive=True会在使用此queue的消费者断开后,自动将queue删除
queue_name = result.method.queuechannel.queue_bind(exchange='logs', # 绑定转发器，收转发器上面的数据queue=queue_name)print(' [*] Waiting for logs. To exit press CTRL+C')def callback(ch, method, properties, body):print(" [x] %r" % body)channel.basic_consume(callback,queue=queue_name,no_ack=True)
channel.start_consuming()consume.py
② 有选择的接收消息 direct: 同fanout一样，指定exchange及routingkey；消费端声明queue断开后不自动销毁，no_ack设置为Flase ，消息不会丢失，再次连接消息重新消费
RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='direct_logs',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,body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()publish.py
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='direct_logs',type='direct')result = channel.queue_declare(exclusive=True)
queue_name = result.method.queueseverities = 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)print(' [*] Waiting for logs. To exit press CTRL+C')def callback(ch, method, properties, body):print(" [x] %r:%r" % (method.routing_key, body))channel.basic_consume(callback,queue=queue_name,no_ack=True)channel.start_consuming()consume.py
③ 更细致的消息过滤 topic:
Although using the direct exchange improved our system, it still has limitations - it can't do routing based on multiple criteria.
In our logging system we might want to subscribe to not only logs based on severity, but also based on the source which emitted the log. You might know this concept from the syslog unix tool, which routes logs based on both severity (info/warn/crit...) and facility (auth/cron/kern...).
That would give us a lot of flexibility - we may want to listen to just critical errors coming from 'cron' but also all logs from 'kern'
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='topic_logs',type='topic')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,body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()publish.py
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='topic_logs',type='topic')result = channel.queue_declare(exclusive=True)
queue_name = result.method.queuebinding_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)print(' [*] Waiting for logs. To exit press CTRL+C')def callback(ch, method, properties, body):print(" [x] %r:%r" % (method.routing_key, body))channel.basic_consume(callback,queue=queue_name,no_ack=True)channel.start_consuming()consume.py
RPC（Remote procedure call ）双向通信
To illustrate how an RPC service could be used we're going to create a simple client class. It's going to expose a method named call which sends an RPC request and blocks until the answer is received:
rpc client：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport pika
import uuid,timeclass FibonacciRpcClient(object):def __init__(self):self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))self.channel = self.connection.channel()result = self.channel.queue_declare(exclusive=True)self.callback_queue = result.method.queueself.channel.basic_consume(self.on_response, #只要收到消息就执行on_responseno_ack=True, #不用ack确认queue=self.callback_queue)def on_response(self, ch, method, props, body):if self.corr_id == props.correlation_id: #验证码核对self.response = bodydef call(self, n):self.response = Noneself.corr_id = str(uuid.uuid4())print(self.corr_id)self.channel.basic_publish(exchange='',routing_key='rpc_queue',properties=pika.BasicProperties(reply_to=self.callback_queue, #发送返回信息的队列namecorrelation_id=self.corr_id, #发送uuid 相当于验证码),body=str(n))while self.response is None:self.connection.process_data_events() #非阻塞版的start_consumingprint("no messages")time.sleep(0.5) #测试return int(self.response)fibonacci_rpc = FibonacciRpcClient() #实例化
print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30) #执行call方法
print(" [.] Got %r" % response)rpc server：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lian
import pika
import timeconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))channel = connection.channel()channel.queue_declare(queue='rpc_queue')def fib(n):if n == 0:return 0elif n == 1:return 1else:return fib(n - 1) + fib(n - 2)def on_request(ch, method, props, body):n = int(body)print(" [.] fib(%s)" % n)response = fib(n)ch.basic_publish(exchange='',routing_key=props.reply_to, #回信息队列名properties=pika.BasicProperties(correlation_id=props.correlation_id),body=str(response))ch.basic_ack(delivery_tag=method.delivery_tag)#channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request,queue='rpc_queue')print(" [x] Awaiting RPC requests")
channel.start_consuming()
转载于:https://www.cnblogs.com/lianzhilei/p/5977545.html
Python开发【十一章】：RabbitMQ队列相关推荐
- Python 第十一章 常用第三方模块
常用第三方模块 除了内建的模块,Python还有大量的第三方模块. 基本上,所有的第三方模块都会在PyPI- The Python Package Index 上注册,只要找到对应模块的名字,即可用p ...
- 路飞学城python开发ftp_路飞学城-Python开发-第二章
'''数据结构: menu = { '北京':{ '海淀':{ '五道口':{ 'soho':{}, '网易':{}, 'google':{} }, '中关村':{ '爱奇艺':{}, '汽车之家': ...
- python第十一章习题
11-1 代码: import unittestdef city_country(city,country):return city.title() + ", " + countr ...
- 第二百九十二节，RabbitMQ多设备消息队列-Python开发
RabbitMQ多设备消息队列-Python开发 首先安装Python开发连接RabbitMQ的API,pika模块 pika模块为第三方模块 对于RabbitMQ来说,生产和消费不再针对内存里的一 ...
- 《Deep Learning With Python second edition》英文版读书笔记：第十一章DL for text: NLP、Transformer、Seq2Seq
文章目录 第十一章:Deep learning for text 11.1 Natural language processing: The bird's eye view 11.2 Preparin ...
- python rabitmq_python RabbitMQ队列使用
原博文 2019-01-17 21:17 − python RabbitMQ队列使用 关于python的queue介绍 关于python的队列,内置的有两种,一种是线程queue,另一种是进程queu ...
- 【正点原子Linux连载】第三十一章 U-Boot顶层Makefile详解 -摘自【正点原子】I.MX6U嵌入式Linux驱动开发指南V1.0
1)实验平台:正点原子阿尔法Linux开发板 2)平台购买地址:https://item.taobao.com/item.htm?id=603672744434 2)全套实验源码+手册+视频下载地址: ...
- 数据结构与算法Python语言实现《Data Structures Algorithms in Python》手写课后答案--第十一章
第十一章 本章叙述了不同平衡树的构造性能问题 习题代码如下(部分代码引用书中源代码,源代码位置目录在第二章答案中介绍) #11.1 #(1,A) # \ # (2,B) # \ # (3,C) # \ ...
- 黑帽python第二版（Black Hat Python 2nd Edition）读书笔记 之 第十一章 攻击性取证
黑帽python第二版(Black Hat Python 2nd Edition)读书笔记 之 第十一章 攻击性取证 文章目录 黑帽python第二版(Black Hat Python 2nd Edi ...
最新文章
- coreldraw的线条怎么变成圆头_别再穿到处撞的小白鞋了，这五款春夏小皮鞋，不管怎么搭配都好看...
- Jmeter实现压力测试(多并发测试)
- python manager详解_python 多进程共享全局变量之Manager()详解
- vue 3.x 中全局配置 axios
- android 原生 电子邮件 应用 发送邮件附带 中文名附件时 附件名称乱码问题解决...
- 解决NetworkOnMainThreadException
- C语言课程设计|通讯录管理系统（含完整代码）
- python数据导出excel_Python方法将DBF文件导出到Excel代码示例
- Linux好用的音乐播放器
- 整蛊系列——使小伙伴的电脑自动关机
- 【洛谷P2000】拯救世界
- 编写程序，创建类Mymath，计算圆的周长和面积以及球的表面积和体积，并编写测试代码，结果均保留两位小数。
- java clh_Java多线程编程CLH锁详解
- 双目九视清哺光仪_岳清江｜坚定信念，普通人也能拥有非凡人生——【管鹏企业家书友会】...
- 使用DOS命令操作MySQL
- 香油和一个生鸡蛋，干咳偏方
- python如何保存excel文件
- Element Plus 表格后端排序
- TortoiseSVN安装注意事项及中文语言包安装
- Linux学习笔记_5_文件目录类指令日期，时间