python操作rabbitmq操作数据
全栈工程师开发手册 (作者:栾鹏)
架构系列文章
##一、RabbitMQ 消息队列介绍
RabbitMQ也是消息队列,那RabbitMQ和之前python的Queue有什么区别么?
py 消息队列:线程 queue(同一进程下线程之间进行交互)进程 Queue(父子进程进行交互 或者 同属于同一进程下的多个子进程进行交互)
如果是两个完全独立的python程序,也是不能用上面两个queue进行交互的,或者和其他语言交互有哪些实现方式呢。
【Disk、Socket、其他中间件】这里中间件不仅可以支持两个程序之间交互,可以支持多个程序,可以维护好多个程序的队列。
像这种公共的中间件有好多成熟的产品:
RabbitMQ
ZeroMQ
ActiveMQ
……
RabbitMQ:erlang语言 开发的。
Python中连接RabbitMQ的模块:pika 、Celery(分布式任务队列) 、haigha
可以维护很多的队列
RabbitMQ 教程官网:http://www.rabbitmq.com/getstarted.html
几个概念说明:
Broker:简单来说就是消息队列服务器实体。
Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列。
Queue:消息队列载体,每个消息都会被投入到一个或多个队列。
Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key:路由关键字,exchange根据这个关键字进行消息投递。
vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离。
producer:消息生产者,就是投递消息的程序。
consumer:消息消费者,就是接受消息的程序。
channel:消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务
##二、RabbitMQ基本示例.
###1、Rabbitmq 安装
ubuntu系统
install rabbitmq-server # 直接搞定
以下centos系统
1)Install Erlang
# For EL5:
rpm -Uvh http://download.fedoraproject.org/pub/epel/5/i386/epel-release-5-4.noarch.rpm# For EL6:
rpm -Uvh http://download.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm# For EL7:
rpm -Uvh http://download.fedoraproject.org/pub/epel/7/x86_64/e/epel-release-7-8.noarch.rpmyum install erlang
2)Install RabbitMQ Server
rpm --import https://www.rabbitmq.com/rabbitmq-release-signing-key.asc
yum install rabbitmq-server-3.6.5-1.noarch.rpm
3)use RabbitMQ Server
chkconfig rabbitmq-server on
service rabbitmq-server stop/start
###2、基本示例
发送端 producer
import pika# 建立一个实例
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost',5672) # 默认端口5672,可不写)
# 声明一个管道,在管道里发消息
channel = connection.channel()
# 在管道里声明queue
channel.queue_declare(queue='hello')
# RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange='',routing_key='hello', # queue名字body='Hello World!') # 消息内容
print(" [x] Sent 'Hello World!'")
connection.close() # 队列关闭
接收端 consumer
import pika
import time# 建立实例
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
# 声明管道
channel = connection.channel()# 为什么又声明了一个‘hello’队列?# 如果确定已经声明了,可以不声明。但是你不知道那个机器先运行,所以要声明两次。
channel.queue_declare(queue='hello')def callback(ch, method, properties, body): # 四个参数为标准格式print(ch, method, properties) # 打印看一下是什么# 管道内存对象 内容相关信息 后面讲print(" [x] Received %r" % body)time.sleep(15)ch.basic_ack(delivery_tag = method.delivery_tag) # 告诉生成者,消息处理完成channel.basic_consume( # 消费消息callback, # 如果收到消息,就调用callback函数来处理消息queue='hello', # 你要从那个队列里收消息# no_ack=True # 写的话,如果接收消息,机器宕机消息就丢了# 一般不写。宕机则生产者检测到发给其他消费者)print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() # 开始消费消息
###3、RabbitMQ 消息分发轮询
上面的只是一个生产者、一个消费者,能不能一个生产者多个消费者呢?
可以上面的例子,多启动几个消费者consumer,看一下消息的接收情况。 采用轮询机制;把消息依次分发假如消费者处理消息需要15秒,如果当机了,那这个消息处理明显还没处理完,怎么处理?
(可以模拟消费端断了,分别注释和不注释 no_ack=True 看一下)
你没给我回复确认,就代表消息没处理完。上面的效果消费端断了就转到另外一个消费端去了,但是生产者怎么知道消费端断了呢?
因为生产者和消费者是通过socket连接的,socket断了,就说明消费端断开了。上面的模式只是依次分发,实际情况是机器配置不一样。怎么设置类似权重的操作?
RabbitMQ怎么办呢,RabbitMQ做了简单的处理就能实现公平的分发。
就是RabbitMQ给消费者发消息的时候检测下消费者里的消息数量,如果超过指定值(比如1条),就不给你发了。
只需要在消费者端,channel.basic_consume前加上就可以了。
channel.basic_qos(prefetch_count=1) # 类似权重,按能力分发,如果有一个消息,就不在给你发
channel.basic_consume( # 消费消息
##三、RabbitMQ 消息持久化(durable、properties)
###1、RabbitMQ 相关命令
rabbitmqctl list_queues # 查看当前queue数量及queue里消息数量
###2、消息持久化
- 如果队列里还有消息,RabbitMQ 服务端宕机了呢?消息还在不在?
把RabbitMQ服务重启,看一下消息在不在。
上面的情况下,宕机了,消息就没了,下面看看如何把消息持久化。 每次声明队列的时候,都加上durable,注意每个队列都得写,客户端、服务端声明的时候都得写。
# 在管道里声明queue
channel.queue_declare(queue='hello2', durable=True)
测试结果发现,只是把队列持久化了,但是队列里的消息没了。
durable的作用只是把队列持久化。离消息持久话还差一步: 发送端发送消息时,加上properties
properties=pika.BasicProperties(delivery_mode=2, # 消息持久化)
发送端 producer
import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost',5672)) # 默认端口5672,可不写
channel = connection.channel()
#声明queue
channel.queue_declare(queue='hello2', durable=True) # 若声明过,则换一个名字#n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange='',routing_key='hello2',body='Hello World!',properties=pika.BasicProperties(delivery_mode=2, # make message persistent))print(" [x] Sent 'Hello World!'")
connection.close()
接收端 consumer
import pika
import timeconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello2', durable=True)def callback(ch, method, properties, body):print(" [x] Received %r" % body)time.sleep(10)ch.basic_ack(delivery_tag = method.delivery_tag) # 告诉生产者,消息处理完成channel.basic_qos(prefetch_count=1) # 类似权重,按能力分发,如果有一个消息,就不在给你发
channel.basic_consume( # 消费消息callback, # 如果收到消息,就调用callbackqueue='hello2',# no_ack=True # 一般不写,处理完接收处理结果。宕机则发给其他消费者)print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
##四、RabbitMQ 广播模式(exchange)
前面的效果都是一对一发,如果做一个广播效果可不可以,这时候就要用到exchange了
exchange必须精确的知道收到的消息要发给谁。exchange的类型决定了怎么处理,
类型有以下几种:
- fanout: 所有绑定到此exchange的queue都可以接收消息
- direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息
- topic: 所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息
###1、fanout 纯广播、all
需要queue和exchange绑定,因为消费者不是和exchange直连的,消费者是连在queue上,queue绑定在exchange上,消费者只会在queu里度消息
|------------------------|| /—— queue <—|—> consumer1
producer —|—exchange1 <bind | \ | \—— queue <—|—> consumer2\-|-exchange2 …… ||------------------------|
发送端 publisher 发布、广播
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 注意:这里是广播,不需要声明queue
channel.exchange_declare(exchange='logs', # 声明广播管道type='fanout')# message = ' '.join(sys.argv[1:]) or "info: Hello World!"
message = "info: Hello World!"
channel.basic_publish(exchange='logs',routing_key='', # 注意此处空,必须有body=message)
print(" [x] Sent %r" % message)
connection.close()
接收端 subscriber 订阅
import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='logs',type='fanout')
# 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
result = channel.queue_declare(exclusive=True)
# 获取随机的queue名字
queue_name = result.method.queue
print("random queuename:", queue_name)channel.queue_bind(exchange='logs', # queue绑定到转发器上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()
注意:广播,是实时的,收不到就没了,消息不会存下来,类似收音机。
###2、direct 有选择的接收消息
接收者可以过滤消息,只收我想要的消息
发送端publisher
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='direct_logs',type='direct')
# 重要程度级别,这里默认定义为 info
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()
接收端subscriber
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.queue
# 获取运行脚本所有的参数
severities = 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()
运行接收端,指定接收级别的参数,例:
python direct_sonsumer.py info warning
python direct_sonsumer.py warning error
###3、topic 更细致的过滤
比如把error中,apache和mysql的分别或取出来
发送端publisher
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()
接收端 subscriber
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()
运行接收端,指定接收哪些消息,例:
python topic_sonsumer.py *.info
python topic_sonsumer.py *.error mysql.*
python topic_sonsumer.py '#' # 接收所有消息# 接收所有的 logs run:
# python receive_logs_topic.py "#"# To receive all logs from the facility "kern":
# python receive_logs_topic.py "kern.*"# Or if you want to hear only about "critical" logs:
# python receive_logs_topic.py "*.critical"# You can create multiple bindings:
# python receive_logs_topic.py "kern.*" "*.critical"# And to emit a log with a routing key "kern.critical" type:
# python emit_log_topic.py "kern.critical" "A critical kernel error"
转载请务必保留此出处:http://blog.csdn.net/fgf00/article/details/52872730
###4、RabbitMQ RPC 实现(Remote procedure call)
不知道你有没有发现,上面的流都是单向的,如果远程的机器执行完返回结果,就实现不了了。
如果返回,这种模式叫什么呢,RPC(远程过程调用),snmp就是典型的RPC
RabbitMQ能不能返回呢,怎么返回呢?既是发送端又是接收端。
但是接收端返回消息怎么返回?可以发送到发过来的queue里么?不可以。 返回时,再建立一个queue,把结果发送新的queue里
为了服务端返回的queue不写死,在客户端给服务端发指令的的时候,同时带一条消息说,你结果返回给哪个queue
RPC client
import pika
import uuid
import 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,queue=self.callback_queue) # 收这个queue的消息def on_response(self, ch, method, props, body): # 必须四个参数# 如果收到的ID和本机生成的相同,则返回的结果就是我想要的指令返回的结果if self.corr_id == props.correlation_id:self.response = bodydef call(self, n):self.response = None # 初始self.response为Noneself.corr_id = str(uuid.uuid4()) # 随机唯一字符串self.channel.basic_publish(exchange='',routing_key='rpc_queue', # 发消息到rpc_queueproperties=pika.BasicProperties( # 消息持久化reply_to = self.callback_queue, # 让服务端命令结果返回到callback_queuecorrelation_id = self.corr_id, # 把随机uuid同时发给服务器),body=str(n))while self.response is None: # 当没有数据,就一直循环# 启动后,on_response函数接到消息,self.response 值就不为空了self.connection.process_data_events() # 非阻塞版的start_consuming()# print("no msg……")# time.sleep(0.5)# 收到消息就调用on_responsereturn int(self.response)if __name__ == '__main__':fibonacci_rpc = FibonacciRpcClient()print(" [x] Requesting fib(7)")response = fibonacci_rpc.call(7)print(" [.] Got %r" % response)
RPC server
import pika
import timedef 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, # 客户端要求返回想用的queue# 返回客户端发过来的correction_id 为了让客户端验证消息一致性properties=pika.BasicProperties(correlation_id = props.correlation_id),body=str(response))ch.basic_ack(delivery_tag = method.delivery_tag) # 任务完成,告诉客户端if __name__ == '__main__':connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))channel = connection.channel()channel.queue_declare(queue='rpc_queue') # 声明一个rpc_queue ,channel.basic_qos(prefetch_count=1)# 在rpc_queue里收消息,收到消息就调用on_requestchannel.basic_consume(on_request, queue='rpc_queue')print(" [x] Awaiting RPC requests")channel.start_consuming()
转载于http://blog.csdn.net/fgf00/article/details/52872730
openstack 默认用的RabbitMQ
参考页面:https://www.zouyesheng.com/rabbitmq.html
python操作rabbitmq操作数据相关推荐
- python操作RabbitMQ
RabbitMQ介绍 RabbitMQ是一个由erlang开发的AMQP(Advanced Message Queue)的开源实现的产品,RabbitMQ是一个消息代理,从"生产者" ...
- Python菜鸟之路:Python基础-Python操作RabbitMQ
RabbitMQ简介 rabbitmq中文翻译的话,主要还是mq字母上:Message Queue,即消息队列的意思.rabbitmq服务类似于mysql.apache服务,只是提供的功能不一样.ra ...
- Python操作 RabbitMQ、Redis、Memcache、SQLAlchemy
Memcached Memcached 是一个高性能的分布式内存对象缓存系统,用于动态Web应用以减轻数据库负载.它通过在内存中缓存数据和对象来减少读取数据库的次数,从而提高动态.数据库驱动网站的速度 ...
- python 操作RabbitMQ
pip install pika使用API操作RabbitMQ基于Queue实现生产者消费者模型View Code 对于RabbitMQ来说,生产和消费不再针对内存里的一个Queue对象,而是某台服务 ...
- python_day10のPython操作 RabbitMQ、Redis、Memcache
Memcached Memcached 是一个高性能的分布式内存对象缓存系统,用于动态Web应用以减轻数据库负载.它通过在内存中缓存数据和对象来减少读取数据库的次数,从而提高动态.数据库驱动网站的速度 ...
- day12 Python操作rabbitmq及pymsql
一.rabbitmq介绍 RabbitMQ简介 AMQP,即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计.消 ...
- 【Python之路Day12】网络篇之Python操作RabbitMQ
基础知识 分布式系统中,我们广泛运用消息中间件进行系统间的数据交换,便于异步解耦.消息中间件这块在我们前面的学习中,是使用python中的queue模块来提供,但这个模块仅限于在本机的内存中使用,假设 ...
- Python之操作RabbitMQ
RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统.他遵循Mozilla Public License开源协议. MQ全称为Message Queue, 消息队列(MQ)是一种应用程序 ...
- python总线 rabbitmq_python - 操作RabbitMQ
介绍 RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统.他遵循Mozilla Public License开源协议. MQ全称为Message Queue, 消息队列(MQ)是一种应 ...
最新文章
- 编写矩形类 计算矩形面积
- weiler-atherton多边形裁剪算法_EAST算法超详细源码解析:数据预处理与标签生成...
- 在几何画板中如何制作圆柱的侧面展开动画_倒计时与时钟演示 | 几何画板
- javaweb项目部署到tomcat之后java文件没有编译
- linux外部命令帮助,Linux的命令帮助
- 配置独立于系统的PYTHON环境
- 三菱模拟量输入与输出程序_初学PLC是学习西门子还是三菱?
- Java文件类boolean isDirectory()方法(带示例)
- 体验微软反间谍软件及恶意软件清除工具
- Sql Server判断某列字段是否为空或判断某列字段长度
- 【开源软件】windows环境下libcurl编译
- h标签在seo中的作用(技术优化)
- 基于LM317的直流稳压电源设计
- macos ntfs插件_NTFS for Mac 助手 - Mac读写NTFS磁盘工具
- 巴克莱银行申请区块链专利改进银行服务
- matlab aic怎么用,AIC信息准则的编程
- 自己写的Excel查询小工具,需要的可以拿走
- 范数和条件数(Norms and Condition Numbers)
- 商业方向的大数据专业_工业大数据应用的三大挑战和五大商业趋势
- 6plus经常显示无服务器,苹果6plus屏幕失灵怎么办?如何解决?
热门文章
- 科大讯飞免切换语音输入,留住更美乡音!
- android自动更新demo,Android程序自动更新功能模块的实现方法【附完整demo源码下载】...
- 网页特效offset、client、scroll系列属性的作用
- 【MySQL快速入门】高级查询:计算函数分组计算
- X264_最简单的视频编码实现(YUV420编码H264)
- 常用编码软件简单使用记录 1 : 自主编码器
- 数据库操作 linq php,.NET_asp.net使用LINQ to SQL连接数据库及SQL操作语句用法分析,本文实例讲述了asp.net使用LINQ t - phpStudy...
- HTML显示波形,CSS3波形loading动画特效
- array in java,ArrayList to Array Conversion in Java
- html读取json换行无效,前端Json换行显示