python第十二天-----RabbitMQ

有一位小伙伴说让我去A站写博客可能会有很多人喜欢，真是搞不懂，北方哪里有卖萌？北方默认状态就是这么萌的！再者说了，这明明就是很专注于技术的博客嘛，能不能严肃点！知不知道什么叫帧？

学习到了数据库的相关操作，真是B了狗了，这个破玩意真是无孔不入啊，从第一次接触到现在一直都对数据库处于抵抗状态！好讨厌！所以最后决定，只写rabbitmq相关的博客，数据库什么的，拉黑！

总的来说10.1学习计划还是很失败的！docker学习进度超出了预期，openstakc学习进度超出了预期，python学习进度也超出了预期，主要是没有吃好，至出去撸了一次串，不够过瘾！后天就要上班了，连上7天，wtf！只要别叫去另一个工作地点就好，后天解决rancher，大后天解决ceph，周一看看有多少小伙伴跑路.....（第一次遇到跑路比我还积极的，而且一次还碰到一群#24）

专注RabbitMQ（pip install pika）

消息队列服务器，可以对业务进行解耦，实现业务缓解消峰，当然保证业务可靠性也有一些，具体的，去百度吧！开发帖里不写运维

1.基本的使用

#!/usr/bin/env python
# ########################## 消费者 ##########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='172.16.5.7'))
channel = connection.channel()                      # 与服务器创建连接

channel.queue_declare(queue='hello')                # 与hello队列关联


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)


channel.basic_consume(callback,
                      queue='hello',
                      no_ack=True)                  # 从hello队列里取出一条消息

channel.start_consuming()

#!/usr/bin/env python
# ######################### 生产者 #########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='172.16.5.7'))
channel = connection.channel()                  # 与服务器创建连接

channel.queue_declare(queue='hello')            # 与hello队列关联

channel.basic_publish(exchange='',              # 向hello队列里放入消息'Hello World!'
                      routing_key='hello',
                      body='Hello World!')
connection.close()

2.在消费者端出现状况时保证消息不丢失，只有收到ack回文后该消息才会被确认消费消除，否则仍然会存在于队列当中，会影响性能，自行取舍

#!/usr/bin/env python
# ########################## 消费者 ##########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='172.16.5.7'))
channel = connection.channel()                      

channel.queue_declare(queue='hello')                


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)    
    ch.basic_ack(delivery_tag=method.delivery_tag)  # 这里

channel.basic_consume(callback,
                      queue='hello',
                      no_ack=False)                  # 这里

channel.start_consuming()

3.生产者在产生消息时可决定是否将此消息持久化到磁盘上，可保证消息队列服务器本身出现问题时消息不丢失

#!/usr/bin/env python
# ######################### 生产者 #########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='172.16.5.7'))
channel = connection.channel()                  

channel.queue_declare(queue='hello', durable=True)            # 这里

channel.basic_publish(exchange='',              
                      routing_key='hello',
                      body='Hello World!',
                      properties = pika.BasicProperties(delivery_mode=2,))  # 这里
connection.close()

4.rmq默认是由消费者个数来各自消费自己对应的任务，为防止某任务消费时间过长卡服务

#!/usr/bin/env python
# ########################## 消费者 ##########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='172.16.5.7'))
channel = connection.channel()

channel.queue_declare(queue='hello')


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)  
    
channel.basic_qos(prefetch_count=1)                 # 谁来谁取任务，一个接一个

channel.basic_consume(callback,
                      queue='hello',
                      no_ack=False)                  

channel.start_consuming()

5.发布订阅（上面的功能也是支持的哦）：发布订阅会将收到的任务消息放入所有订阅者，而普通的被消费一次就gg思密达了

此时就要用到exchange功能了，由他来关联对应的队列，关联方式有三种（完全发布fanout，关键字发布direct，模糊匹配关键字发布topic）

完全发布模式

#!/usr/bin/env python
# ########################## 消费者 ##########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='172.16.5.7'))
channel = connection.channel()

channel.exchange_declare(exchange='logs',               # exchange的名称
                         type='fanout')                 # exchange的类型，完全发布

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue                        # 随机生成一个队列名（对于我这种起名困哪户简直是福音...）

channel.queue_bind(exchange='logs',
                   queue=queue_name)                    # 将刚才的队列绑定到exchange中


def callback(ch, method, properties, body):
    print(" [x] %r" % body)

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

#!/usr/bin/env python
# ######################### 生产者 #########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.16.5.7'))
channel = connection.channel()
channel.exchange_declare(exchange='logs', type='fanout')
message = '456'
channel.basic_publish(exchange='logs', routing_key='', body=message)        # 向exchange里发布消息
print('[x] send %s' % message)
connection.close()

关键字模式

#!/usr/bin/env python
# ######################### 消费者 #########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.16.5.7'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', type='direct')         # 关键字模式
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = ['info', 'error', 'warning']               # 关键字3个

for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)


def call_back(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))

channel.basic_consume(call_back,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

#!/usr/bin/env python
# ######################### 消费者 #########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.16.5.7'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = ['error']                              # 关键字1个

for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)


def call_back(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))

channel.basic_consume(call_back,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

#!/usr/bin/env python
# ########################## 生产者 ##########################
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.16.5.7'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',
                         type='direct')

severity = 'info'                                               # 关键字'info'只有一个队列收取
# severity = 'error'                                            # 关键字'error'都收取
message = '123'
channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

模糊匹配模式（我不用....）

# 表示可以匹配 0 个 或 多个 单词

*  表示只能匹配 一个 单词

abc.aaaaa　　　　abc.# 不匹配

abc.aaaaa　　　　abc.* 匹配

想要吃好吃的食物