0

0

利用Python学习RabbitMQ消息队列

php中文网

php中文网

发布时间:2016-06-10 15:07:06

|

1157人浏览过

|

来源于php中文网

原创

rabbitmq可以当做一个消息代理,它的核心原理非常简单:即接收和发送消息,可以把它想象成一个邮局:我们把信件放入邮箱,邮递员就会把信件投递到你的收件人处,rabbitmq就是一个邮箱、邮局、投递员功能综合体,整个过程就是:邮箱接收信件,邮局转发信件,投递员投递信件到达收件人处。

RabbitMQ和邮局的主要区别就是RabbitMQ接收、存储和发送的是二进制数据----消息。

rabbitmq基本管理命令:

一步启动Erlang node和Rabbit应用:sudo rabbitmq-server

在后台启动Rabbit node:sudo rabbitmq-server -detached

立即学习Python免费学习笔记(深入)”;

关闭整个节点(包括应用):sudo rabbitmqctl stop

add_user  
delete_user 
change_password  
list_users
add_vhost 
delete_vhost 
list_vhosts
set_permissions [-p ]    
clear_permissions [-p ] 
list_permissions [-p ]
list_user_permissions 
list_queues [-p ] [ ...]
list_exchanges [-p ] [ ...]
list_bindings [-p ]
list_connections [ ...]

Demo:

producer.py

 #!/usr/bin/env python
 # -*- coding: utf_ -*-
 # Date: 年月日
 # Author:蔚蓝行
 # 博客 http://www.cnblogs.com/duanv/
 import pika
 import sys
 #创建连接connection到localhost
 con = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
 #创建虚拟连接channel
 cha = con.channel()
 #创建队列anheng,durable参数为真时,队列将持久化;exclusive为真时,建立临时队列
 result=cha.queue_declare(queue='anheng',durable=True,exclusive=False)
 #创建名为yanfa,类型为fanout的exchange,其他类型还有direct和topic,如果指定durable为真,exchange将持久化
 cha.exchange_declare(durable=False,
           exchange='yanfa',
           type='direct',)
 #绑定exchange和queue,result.method.queue获取的是队列名称
 cha.queue_bind(exchange='yanfa', 
        queue=result.method.queue,
        routing_key='',) 
 #公平分发,使每个consumer在同一时间最多处理一个message,收到ack前,不会分配新的message
 cha.basic_qos(prefetch_count=)
 #发送信息到队列‘anheng'
 message = ' '.join(sys.argv[:])
 #消息持久化指定delivery_mode=;
 cha.basic_publish(exchange='',
          routing_key='anheng',
          body=message,
          properties=pika.BasicProperties(
           delivery_mode = ,
         ))
 print '[x] Sent %r' % (message,)
 #关闭连接
 con.close()

consumer.py

 #!/usr/bin/env python
 # -*- coding: utf_ -*-
 # Date: 年月日
 # Author:蔚蓝行
 # 博客 http://www.cnblogs.com/duanv/
 import pika
 #建立连接connection到localhost
 con = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
 #创建虚拟连接channel
 cha = con.channel()
 #创建队列anheng
 result=cha.queue_declare(queue='anheng',durable=True)
 #创建名为yanfa,类型为fanout的交换机,其他类型还有direct和topic
 cha.exchange_declare(durable=False,
           exchange='yanfa', 
           type='direct',)
 #绑定exchange和queue,result.method.queue获取的是队列名称
 cha.queue_bind(exchange='yanfa',
        queue=result.method.queue,
        routing_key='',)
 #公平分发,使每个consumer在同一时间最多处理一个message,收到ack前,不会分配新的message
 cha.basic_qos(prefetch_count=)
 print ' [*] Waiting for messages. To exit press CTRL+C'
 #定义回调函数
 def callback(ch, method, properties, body):
   print " [x] Received %r" % (body,)
   ch.basic_ack(delivery_tag = method.delivery_tag)
 cha.basic_consume(callback,
          queue='anheng',
          no_ack=False,)
 cha.start_consuming()

一、概念:

Connection: 一个TCP的连接。Producer和Consumer都是通过TCP连接到RabbitMQ Server的。程序的起始处就是建立这个TCP连接。

Channels: 虚拟连接。建立在上述的TCP连接中。数据流动都是在Channel中进行的。一般情况是程序起始建立TCP连接,第二步就是建立这个Channel。

二、队列:

首先建立一个Connection,然后建立Channels,在channel上建立队列

建立时指定durable参数为真,队列将持久化;指定exclusive为真,队列为临时队列,关闭consumer后该队列将不再存在,一般情况下建立临时队列并不指定队列名称,rabbitmq将随机起名,通过result.method.queue来获取队列名:

result = channel.queue_declare(exclusive=True)

result.method.queue

区别:durable是队列持久化与否,如果为真,队列将在rabbitmq服务重启后仍存在,如果为假,rabbitmq服务重启前不会消失,与consumer关闭与否无关;

而exclusive是建立临时队列,当consumer关闭后,该队列就会被删除

三、exchange和bind

Exchange中durable参数指定exchange是否持久化,exchange参数指定exchange名称,type指定exchange类型。Exchange类型有direct,fanout和topic。

Bind是将exchange与queue进行关联,exchange参数和queue参数分别指定要进行bind的exchange和queue,routing_key为可选参数。

Exchange的三种模式:

Direct:

任何发送到Direct Exchange的消息都会被转发到routing_key中指定的Queue

1.一般情况可以使用rabbitMQ自带的Exchange:””(该Exchange的名字为空字符串);

2.这种模式下不需要将Exchange进行任何绑定(bind)操作;

3.消息传递时需要一个“routing_key”,可以简单的理解为要发送到的队列名字;

4.如果vhost中不存在routing_key中指定的队列名,则该消息会被抛弃。

Demo中虽然声明了一个exchange='yanfa'和queue='anheng'的bind,但是在后面发送消息时并没有使用该exchange和bind,而是采用了direct的模式,没有指定exchange,而是指定了routing_key的名称为队列名,消息将发送到指定队列。

如果一个exchange 声明为direct,并且bind中指定了routing_key,那么发送消息时需要同时指明该exchange和routing_key.

晓语台
晓语台

晓语台,是一款AI文本创作产品。创作能力主要围绕营销文本的AI创作,晓语台覆盖了品牌与市调、商业媒体、社交媒体、搜索营销、数字广告、职场办公共六类全营销文本

下载

Fanout:

任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上

1.可以理解为路由表的模式

2.这种模式不需要routing_key

3.这种模式需要提前将Exchange与Queue进行绑定,一个Exchange可以绑定多个Queue,一个Queue可以同多个Exchange进行绑定。

4.如果接受到消息的Exchange没有与任何Queue绑定,则消息会被抛弃。

Demo中创建了一个将一个exchange和一个queue进行fanout类型的bind.但是发送信息时没有用到它,如果要用到它,只要在发送消息时指定该exchange的名称即可,该exchange就会将消息发送到所有和它bind的队列中。在fanout模式下,指定的routing_key是无效的 。

Topic:

任何发送到Topic Exchange的消息都会被转发到所有关心routing_key中指定话题的Queue上

1.这种模式较为复杂,简单来说,就是每个队列都有其关心的主题,所有的消息都带有一个“标题”(routing_key),Exchange会将消息转发到所有关注主题能与routing_key模糊匹配的队列。

2.这种模式需要routing_key,也许要提前绑定Exchange与Queue。

3.在进行绑定时,要提供一个该队列关心的主题,如“#.log.#”表示该队列关心所有涉及log的消息(一个routing_key为”MQ.log.error”的消息会被转发到该队列)。

4.“#”表示0个或若干个关键字,“*”表示一个关键字。如“log.*”能与“log.warn”匹配,无法与“log.warn.timeout”匹配;但是“log.#”能与上述两者匹配。

5.同样,如果Exchange没有发现能够与routing_key匹配的Queue,则会抛弃此消息。

四、任务分发

1.Rabbitmq的任务是循环分发的,如果开启两个consumer,producer发送的信息是轮流发送到两个consume的。

2.在producer端使用cha.basic_publish()来发送消息,其中body参数就是要发送的消息,properties=pika.BasicProperties(delivery_mode = 2,)启用消息持久化,可以防止RabbitMQ Server 重启或者crash引起的数据丢失。

3.在接收端使用cha.basic_consume()无限循环监听,如果设置no-ack参数为真,每次Consumer接到数据后,而不管是否处理完成,RabbitMQ Server会立即把这个Message标记为完成,然后从queue中删除了。为了保证数据不被丢失,RabbitMQ支持消息确认机制,即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到,那么我们不能采用no-ack。而应该是在处理完数据后发送ack。

在处理数据后发送的ack,就是告诉RabbitMQ数据已经被接收,处理完成,RabbitMQ可以去安全的删除它了。如果Consumer退出了但是没有发送ack,那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。

这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说,RabbitMQ给了Consumer足够长的时间来做数据处理。

Demo的callback方法中ch.basic_ack(delivery_tag = method.delivery_tag)告诉rabbitmq消息已经正确处理。如果没有这条代码,Consumer退出时,Message会重新分发。然后RabbitMQ会占用越来越多的内存,由于RabbitMQ会长时间运行,因此这个“内存泄漏”是致命的。去调试这种错误,可以通过一下命令打印un-acked Messages:

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

4.公平分发:设置cha.basic_qos(prefetch_count=1),这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说,在接收到该Consumer的ack前,他它不会将新的Message分发给它。

五、注意:

生产者和消费者都应该声明建立队列,网上教程上说第二次创建如果参数和第一次不一样,那么该操作虽然成功,但是queue的属性并不会被修改。

可能因为版本问题,在我的测试中如果第二次声明建立的队列属性和第一次不完全相同,将报类似这种错406, "PRECONDITION_FAILED - parameters for queue 'anheng' in vhost '/' not equivalent"

如果是exchange第二次创建属性不同,将报这种错406, "PRECONDITION_FAILED - cannot redeclare exchange 'yanfa' in vhost '/' with different type, durable, internal or autodelete value"

如果第一次声明建立队列也出现这个错误,说明之前存在名字相同的队列且本次声明的某些属性和之前声明不同,可通过命令sudo rabbitmqctl list_queues查看当前有哪些队列。解决方法是声明建立另一名称的队列或删除原有队列,如果原有队列是非持久化的,可通过重启rabbitmq服务删除原有队列,如果原有队列是持久化的,只能删除它所在的vhost,然后再重建vhost,再设置vhost的权限(先确认该vhost中没有其他有用队列)。

sudo rabbitmqctl delete_vhost /
sudo rabbitmqctl add_vhost /
sudo rabbitmqctl set_permissions -p / username '.*' '.*' '.*'

以上内容是小编给大家介绍的利用Python学习RabbitMQ消息队列,希望大家喜欢。

相关文章

python速学教程(入门到精通)
python速学教程(入门到精通)

python怎么学习?python怎么入门?python在哪学?python怎么学才快?不用担心,这里为大家提供了python速学教程(入门到精通),有需要的小伙伴保存下载就能学习啦!

下载

本站声明:本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn

相关专题

更多
Word 字间距调整方法汇总
Word 字间距调整方法汇总

本专题整合了Word字间距调整方法,阅读下面的文章了解更详细操作。

2

2025.12.24

任务管理器教程
任务管理器教程

本专题整合了任务管理器相关教程,阅读下面的文章了解更多详细操作。

2

2025.12.24

AppleID格式
AppleID格式

本专题整合了AppleID相关内容,阅读专题下面的文章了解更多详细教程。

0

2025.12.24

csgo视频观看入口合集
csgo视频观看入口合集

本专题整合了csgo观看入口合集,阅读下面的文章了知道更多入口地址。

29

2025.12.24

yandex外贸入口合集
yandex外贸入口合集

本专题汇总了yandex外贸入口地址,阅读下面的文章了解更多内容。

58

2025.12.24

添加脚注通用方法
添加脚注通用方法

本专题整合了添加脚注方法合集,阅读专题下面的文章了解更多内容。

1

2025.12.24

重启电脑教程汇总
重启电脑教程汇总

本专题整合了重启电脑操作教程,阅读下面的文章了解更多详细教程。

3

2025.12.24

纸张尺寸汇总
纸张尺寸汇总

本专题整合了纸张尺寸相关内容,阅读专题下面的文章了解更多内容。

5

2025.12.24

Java Spring Boot 微服务实战
Java Spring Boot 微服务实战

本专题深入讲解 Java Spring Boot 在微服务架构中的应用,内容涵盖服务注册与发现、REST API开发、配置中心、负载均衡、熔断与限流、日志与监控。通过实际项目案例(如电商订单系统),帮助开发者掌握 从单体应用迁移到高可用微服务系统的完整流程与实战能力。

1

2025.12.24

热门下载

更多
网站特效
/
网站源码
/
网站素材
/
前端模板

精品课程

更多
相关推荐
/
热门推荐
/
最新课程
关于我们 免责申明 举报中心 意见反馈 讲师合作 广告合作 最新更新
php中文网:公益在线php培训,帮助PHP学习者快速成长!
关注服务号 技术交流群
PHP中文网订阅号
每天精选资源文章推送

Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号