首页 > 代码库 > python第十二天-----RabbitMQ
python第十二天-----RabbitMQ
有一位小伙伴说让我去A站写博客可能会有很多人喜欢，真是搞不懂，北方哪里有卖萌？北方默认状态就是这么萌的！
再者说了，这明明就是很专注于技术的博客嘛，能不能严肃点！
知不知道什么叫帧？
学习到了数据库的相关操作，真是B了狗了，这个破玩意真是无孔不入啊，从第一次接触到现在一直都对数据库处于抵抗状态！好讨厌！所以最后决定，只写rabbitmq相关的博客，数据库什么的，拉黑！
总的来说10.1学习计划还是很失败的！docker学习进度超出了预期，openstakc学习进度超出了预期，python学习进度也超出了预期，主要是没有吃好，至出去撸了一次串，不够过瘾！后天就要上班了，连上7天，wtf！只要别叫去另一个工作地点就好，后天解决rancher，大后天解决ceph，周一看看有多少小伙伴跑路.....（第一次遇到跑路比我还积极的，而且一次还碰到一群#24）
专注RabbitMQ（pip install pika）
消息队列服务器，可以对业务进行解耦，实现业务缓解消峰，当然保证业务可靠性也有一些，具体的，去百度吧！开发帖里不写运维
1.基本的使用
1 #!/usr/bin/env python 2 # ########################## 消费者 ########################## 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host=‘172.16.5.7‘)) 7 channel = connection.channel() # 与服务器创建连接 8 9 channel.queue_declare(queue=‘hello‘) # 与hello队列关联10 11 12 def callback(ch, method, properties, body):13 print(" [x] Received %r" % body)14 15 16 channel.basic_consume(callback,17 queue=‘hello‘,18 no_ack=True) # 从hello队列里取出一条消息19 20 channel.start_consuming()
1 #!/usr/bin/env python 2 # ######################### 生产者 ######################### 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host=‘172.16.5.7‘)) 7 channel = connection.channel() # 与服务器创建连接 8 9 channel.queue_declare(queue=‘hello‘) # 与hello队列关联10 11 channel.basic_publish(exchange=‘‘, # 向hello队列里放入消息‘Hello World!‘12 routing_key=‘hello‘,13 body=‘Hello World!‘)14 connection.close()
2.在消费者端出现状况时保证消息不丢失，只有收到ack回文后该消息才会被确认消费消除，否则仍然会存在于队列当中，会影响性能，自行取舍
1 #!/usr/bin/env python 2 # ########################## 消费者 ########################## 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host=‘172.16.5.7‘)) 7 channel = connection.channel() 8 9 channel.queue_declare(queue=‘hello‘) 10 11 12 def callback(ch, method, properties, body):13 print(" [x] Received %r" % body) 14 ch.basic_ack(delivery_tag=method.delivery_tag) # 这里15 16 channel.basic_consume(callback,17 queue=‘hello‘,18 no_ack=False) # 这里19 20 channel.start_consuming()
3.生产者在产生消息时可决定是否将此消息持久化到磁盘上，可保证消息队列服务器本身出现问题时消息不丢失
1 #!/usr/bin/env python 2 # ######################### 生产者 ######################### 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host=‘172.16.5.7‘)) 7 channel = connection.channel() 8 9 channel.queue_declare(queue=‘hello‘, durable=True) # 这里10 11 channel.basic_publish(exchange=‘‘, 12 routing_key=‘hello‘,13 body=‘Hello World!‘,14 properties = pika.BasicProperties(delivery_mode=2,)) # 这里15 connection.close()
4.rmq默认是由消费者个数来各自消费自己对应的任务，为防止某任务消费时间过长卡服务
1 #!/usr/bin/env python 2 # ########################## 消费者 ########################## 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host=‘172.16.5.7‘)) 7 channel = connection.channel() 8 9 channel.queue_declare(queue=‘hello‘)10 11 12 def callback(ch, method, properties, body):13 print(" [x] Received %r" % body)14 ch.basic_ack(delivery_tag=method.delivery_tag) 15 16 channel.basic_qos(prefetch_count=1) # 谁来谁取任务，一个接一个17 18 channel.basic_consume(callback,19 queue=‘hello‘,20 no_ack=False) 21 22 channel.start_consuming()
5.发布订阅（上面的功能也是支持的哦）：发布订阅会将收到的任务消息放入所有订阅者，而普通的被消费一次就gg思密达了
此时就要用到exchange功能了，由他来关联对应的队列，关联方式有三种（完全发布fanout，关键字发布direct，模糊匹配关键字发布topic）
完全发布模式
1 #!/usr/bin/env python 2 # ########################## 消费者 ########################## 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host=‘172.16.5.7‘)) 7 channel = connection.channel() 8 9 channel.exchange_declare(exchange=‘logs‘, # exchange的名称10 type=‘fanout‘) # exchange的类型，完全发布11 12 result = channel.queue_declare(exclusive=True)13 queue_name = result.method.queue # 随机生成一个队列名（对于我这种起名困哪户简直是福音...）14 15 channel.queue_bind(exchange=‘logs‘,16 queue=queue_name) # 将刚才的队列绑定到exchange中17 18 19 def callback(ch, method, properties, body):20 print(" [x] %r" % body)21 22 channel.basic_consume(callback,23 queue=queue_name,24 no_ack=True)25 26 channel.start_consuming()
1 #!/usr/bin/env python 2 # ######################### 生产者 ######################### 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘172.16.5.7‘)) 6 channel = connection.channel() 7 channel.exchange_declare(exchange=‘logs‘, type=‘fanout‘) 8 message = ‘456‘ 9 channel.basic_publish(exchange=‘logs‘, routing_key=‘‘, body=message) # 向exchange里发布消息10 print(‘[x] send %s‘ % message)11 connection.close()
关键字模式
1 #!/usr/bin/env python 2 # ######################### 消费者 ######################### 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘172.16.5.7‘)) 6 channel = connection.channel() 7 channel.exchange_declare(exchange=‘direct_logs‘, type=‘direct‘) # 关键字模式 8 result = channel.queue_declare(exclusive=True) 9 queue_name = result.method.queue10 severities = [‘info‘, ‘error‘, ‘warning‘] # 关键字3个11 12 for severity in severities:13 channel.queue_bind(exchange=‘direct_logs‘,14 queue=queue_name,15 routing_key=severity)16 17 18 def call_back(ch, method, properties, body):19 print(" [x] %r:%r" % (method.routing_key, body))20 21 channel.basic_consume(call_back,22 queue=queue_name,23 no_ack=True)24 25 channel.start_consuming()
1 #!/usr/bin/env python 2 # ######################### 消费者 ######################### 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘172.16.5.7‘)) 6 channel = connection.channel() 7 channel.exchange_declare(exchange=‘direct_logs‘, type=‘direct‘) 8 result = channel.queue_declare(exclusive=True) 9 queue_name = result.method.queue10 severities = [‘error‘] # 关键字1个11 12 for severity in severities:13 channel.queue_bind(exchange=‘direct_logs‘,14 queue=queue_name,15 routing_key=severity)16 17 18 def call_back(ch, method, properties, body):19 print(" [x] %r:%r" % (method.routing_key, body))20 21 channel.basic_consume(call_back,22 queue=queue_name,23 no_ack=True)24 25 channel.start_consuming()
1 #!/usr/bin/env python 2 # ########################## 生产者 ########################## 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘172.16.5.7‘)) 6 channel = connection.channel() 7 8 channel.exchange_declare(exchange=‘direct_logs‘, 9 type=‘direct‘)10 11 severity = ‘info‘ # 关键字‘info‘只有一个队列收取12 # severity = ‘error‘ # 关键字‘error‘都收取13 message = ‘123‘14 channel.basic_publish(exchange=‘direct_logs‘,15 routing_key=severity,16 body=message)17 print(" [x] Sent %r:%r" % (severity, message))18 connection.close()
模糊匹配模式（我不用....）
# 表示可以匹配 0 个 或 多个 单词
* 表示只能匹配 一个 单词
abc.aaaaa abc.# 不匹配
abc.aaaaa abc.* 匹配
想要吃好吃的食物
python第十二天-----RabbitMQ
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。