python网络篇【第十二篇】RabbitMQ

RabbitMQ

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

RabbitMQ Server： 是一种传入服务。 它的角色是维护一条从生产者(Producer) 到 消费者(Consumer)的路线，从而保证数据能够按照指定的方式进行传入。但是也并不是100%的保证数据完整，对于普通的应用来说，应该是足够的。当然对于要求可靠性、完整性绝对的场景，可以再走一层数据一致性的guard， 就可以保证了。

RabbitMQ安装：

安装配置epel源
   $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm
 
安装erlang
   $ yum -y install erlang
 
安装RabbitMQ
   $ yum -y install rabbitmq-server

启动关闭：service rabbitmq-server start/stop

安装API：

pip3 install pika

基于Queue实现生产者消费者模型：

import queue
import threading
m=queue.Queue()

def producer(i):
    while True:
        m.put(i)

def consumer(i):
    while True:
        g=m.get(i)
        print(g)

for i in range(12):
    t=threading.Thread(target=producer,args=(i,))
    t.start()

for i in range(10):
    t=threading.Thread(target=consumer,args=(i,))
    t.start()

对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。相当于C/S架构了。

生产者：

#####生产者：

import pika
connection=pika.BlockingConnection(
    pika.ConnectionParameters(host="192.168.1.129")   #建立TCP连接，端口不写代表默认
)

chanel=connection.channel()  #虚拟连接，建立在上面的TCP连接基础之上
##为什么使用Channel,而不用TCP连接?
#因为对于OS来说,建立和关闭TCP连接是有代价的,尤其是频繁的建立和关闭. 而且TCP的连接数默认在系统内核中也有限制, 这也限制了系统处理高并发的能力.
#但是,如果在TCP连接中建立 Channel是没有代价的,对于Procuder或者Consumer来讲,可以并发的使用多个channel来进行publish或者receive.

chanel.queue_declare(queue="Tom")    #创建一个，队列名字为Tom

chanel.basic_publish(
    exchange="",          #信息是不能直接发送到队列的，它需要发送到交换机(exchange),下面会谈这个,此处使用一个空字符串来标识.
    routing_key="Tom",   #必须指定队列的的名称
    body="Hello Tom" , # 所要发送的信息
)

print("Send OK")
connection.close()   #关闭连接

消费者：

import pika
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.1.129')
)
channel=connection.channel()
channel.queue_declare("Tom")
def callback(ch,method,properties,body):
    '''
    回调方法，当我们获取到信息的时候，Pika库就会调用此函数，这个回调函数会将接收到的信息内容输出到屏幕上
    :param ch: 虚拟通道
    :param method: 方法
    :param properties: 信息属性
    :param body: 接收到的信息
    :return:
    '''
    print("Received:%r"%body)

while True:
    channel.basic_consume(
        #需要告诉RabbitMQ这个灰调函数将会从名为“Tom”的队列中接收信息：
        callback,
        queue="Tom",
        no_ack=True,
    )
    print("Wating for message ,To exit press CTRL+C")

    channel.start_consuming()

1. 消息确认

避免信息丢失，RabbitMQ提供了消息响应（acknowledgments）。消费者会通过一个ack（响应），告诉RabbitMQ已经收到并处理了某条消息，然后RabbitMQ就会释放并删除这条消息。如果消费者（consumer）挂掉了，没有发送响应，RabbitMQ就会认为消息没有被完全处理，然后重新发送给其他消费者（consumer）。这样，及时工作者（workers）偶尔的挂掉，也不会丢失消息。

消息是没有超时这个概念的；当工作者与它断开连的时候，RabbitMQ会重新发送消息。这样在处理一个耗时非常长的消息任务的时候就不会出问题了。

消息响应：默认是开启的 no_ack， 在上面的例子中，我们把标识给置为True(关闭了)。开启后，完成一个任务后，会发送一个响应。

no-ack ＝ False，如果消费者遇到情况(its channel is closed, connection is closed, or TCP connection is lost)挂掉了，那么，RabbitMQ会重新将该任务添加到队列中

import pika
import time
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.1.129')
)
channel=connection.channel()
channel.queue_declare("Tom")
def callback(ch,method,properties,body):
    '''
    回调方法，当我们获取到信息的时候，Pika库就会调用此函数，这个回调函数会将接收到的信息内容输出到屏幕上
    :param ch: 虚拟通道
    :param method: 方法
    :param properties: 信息属性
    :param body: 接收到的信息
    :return:
    '''
    print("Received:%r"%body)
    time.sleep(20)
    print("ok")
    ch.basic_ack(delivery_tag=method.delivery_tag)   #信息不丢失

while True:
    channel.basic_consume(
        #需要告诉RabbitMQ这个灰调函数将会从名为“Tom”的队列中接收信息：
        callback,
        queue="Tom",
        no_ack=False,   #这个地方改为False
    )
    print("Wating for message ,To exit press CTRL+C")

    channel.start_consuming()

2. 消息持久化

有个参数durable， 默认情况下，如果没有显式告诉RabbitMQ这条消息需要持久化，那么它(rabbitmq)在自己退出或者崩溃的时候，将会丢失所有队列和消息。

为了确保不丢失，需要注意：

必须把队列和消息设置为持久化。

#####生产者#####

import pika
connection=pika.BlockingConnection(
    pika.ConnectionParameters(host="192.168.1.129")
)
channel=connection.channel()
channel.queue_declare(queue="Tom")

channel.basic_publish(
    exchange="",
    routing_key="Tom",
    body="Hello Tom",
    properties=pika.BasicProperties(delivery_mode=2)   #持久化
)
print("[x] Sent 'Hello Tom!'")
connection.close()

#####消费者######
import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.1.129')
)

channel = connection.channel()
channel.queue_declare('Tom')  #使用queue_declare创建一个队列——我们可以运行这个命令很多次，但是只有一个队列会被创建。

def callback(ch, method, properties, body):
    '''
    回调方法,当我们获取到消息的时候，Pika库就会调用此回调函数。这个回调函数会将接收到的消息内容输出到屏幕上。
    :param ch:  虚拟通道
    :param method: 方法
    :param properties: 消息属性
    :param body:  消息主体
    :return:
    '''
    print('Received: %r' %body)
    ch.basic_ack(delivery_tag=method.delivery_tag)

while True:
    channel.basic_consume(    #需要告诉RabbitMQ这个回调函数将会从名为"Tom队列中接收消息：
        callback,
        queue='Tom',
        no_ack=False    #no_ack 置为False
    )

    print('Wating for messages. To exit press CTRL+C')

    channel.start_consuming()

3、消息获取顺序

默认消息队列里的数据是按照顺序被消费者拿走，例如：消费者1 去队列中获取 奇数 序列的任务，消费者1去队列中获取 偶数 序列的任务。

channel.basic_qos(prefetch_count=1) 表示谁来谁取，不再按照奇偶数排列

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


################################消费者###########################################
import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.1.129')
)

channel = connection.channel()
channel.queue_declare('Tom')  #使用queue_declare创建一个队列——我们可以运行这个命令很多次，但是只有一个队列会被创建。

def callback(ch, method, properties, body):
    '''
    回调方法,当我们获取到消息的时候，Pika库就会调用此回调函数。这个回调函数会将接收到的消息内容输出到屏幕上。
    :param ch:  虚拟通道
    :param method: 方法
    :param properties: 消息属性
    :param body:  消息主体
    :return:
    '''
    print('Received: %r' %body)
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)    #设置 prefetch_count=1，告诉RabbitMQ，在同一时刻，不要发送超过一条消息给一个worker，直到它处理了上一条消息并且做出了响应。

while True:
    channel.basic_consume(    #需要告诉RabbitMQ这个回调函数将会从名为"Tom"的队列中接收消息：
        callback,
        queue='Tom',
        no_ack=False
    )

    print('Wating for messages. To exit press CTRL+C')

    channel.start_consuming()

4. 发布/订阅

分发一个消息给多个消费者(Consumers), 这种模式，称为"发布/订阅" 

发布者只需要把消息发送给exchange。exchange一边从发布者放接受消息，一边推送到队列。

exchange必须知道如何处理它接收到的消息，是应该推送到指定的队列还是是多个队列，或者是直接忽略消息。这些规则是通过交换机类型（exchange type）来定义的。

exchange的几个类型：

　　direct: 直连, 通过binding key的完全匹配来传递消息到相应的队列中

　　topic：主题交换机，exchange将传入的”路由值“ 和”关键字“进行匹配，匹配成功，则将数据发送到指定队列。

• # 表示可以匹配0个 或者多个单词
• * 表示只能匹配一个单词

发送者路由值              队列中
old.boy.python          old.*  -- 不匹配
old.boy.python          old.#  -- 匹配

　　fanout：扇形交换机， 把消息发送给和他关联的所有的队列

fanout，绑定(binding)模式

exchange type = fanout

import pika  #导入模块

connection = pika.BlockingConnection(
        pika.ConnectionParameters(host='192.168.1.129') #端口如果是默认的话,不用写
        )   #建立TCP连接

channel = connection.channel()  #虚拟连接,建立在上面的TCP连接基础上

channel.exchange_declare(exchange="logs",
                         type="fanout")   #使用fanout类型

message="Hello,Tom and jerry!"
channel.basic_publish(
    exchange="logs",     #消息是不能直接发送到队列的,它需要发送到交换机(exchange),exchange值必须为定义好的exchange值
    routing_key="",
    body=message,    #所要发送的信息
)

print(" [x] Sent %s"%message)
connection.close()        #关闭连接

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


# fanout
################################消费者###########################################
import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.1.129')
)

channel = connection.channel()
channel.exchange_declare(
    exchange='logs',
    type='fanout'
)

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

channel.queue_bind(
    exchange='logs',
    queue=queue_name
)
print('Wating for messages. To exit press CTRL+C')

def callback(ch, method, properties, body):
    '''
    回调方法,当我们获取到消息的时候，Pika库就会调用此回调函数。这个回调函数会将接收到的消息内容输出到屏幕上。
    :param ch:  虚拟通道
    :param method: 方法
    :param properties: 消息属性
    :param body:  消息主体
    :return:
    '''
    print('Received: %r' %body)
    ch.basic_ack(delivery_tag=method.delivery_tag)


while True:
    channel.basic_consume(    #需要告诉RabbitMQ这个回调函数将会从名为"hello"的队列中接收消息：
        callback,
        queue=queue_name,
        no_ack=False
    )
    channel.start_consuming()

direct，多个绑定（关键字)

多个队列使用相同的绑定键是合法的。上图这个例子中，我们可以添加一个X和Q1之间的绑定，使用black绑定键。这样一来，直连交换机就和扇型交换机的行为一样，会将消息广播到所有匹配的队列。带有black路由键的消息会同时发送到Q1和Q2。

import pika
import sys

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

channel.exchange_declare(exchange='TOM',
                         type='direct')   #使用direct类型

severity = 'info'   #如果生产者发送一个info消息的话,两个消费者都能收到; 如果发送一个error级别的消息,只有 error级别的能收到..
message ="info,123"
channel.basic_publish(exchange='TOM',
                      routing_key=severity,
                      body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

import pika
import sys

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

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

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

severities = ["error","info","warning"]     #日志级别定义为info
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]
" % sys.argv[0])
    sys.exit(1)

for severity in severities:       #递归绑定到哪队
    channel.queue_bind(exchange='TOM',
                       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))
    ch.basic_ack(delivery_tag=method.delivery_tag)
while True:
    channel.basic_consume(callback,
                          queue=queue_name,
                          no_ack=False)

channel.start_consuming()

import pika
import sys

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

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

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

severities = ["error",]     #日志级别定义为info
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]
" % sys.argv[0])
    sys.exit(1)

for severity in severities:       #递归绑定到哪队
    channel.queue_bind(exchange='TOM',
                       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))
    ch.basic_ack(delivery_tag=method.delivery_tag)
while True:
    channel.basic_consume(callback,
                          queue=queue_name,
                          no_ack=False)

channel.start_consuming()

Topic（模糊匹配）

在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。

• # 表示可以匹配 0 个 或 多个 单词
• *  表示只能匹配 一个 单词

import pika
import sys

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

channel.exchange_declare(exchange='hello',type='topic')   #使用direct类型

severity = 'info.ccc'   #如果生产者发送一个info消息的话,两个消费者都能收到; 如果发送一个error级别的消息,只有 error级别的能收到..
message ="info.com"
channel.basic_publish(exchange='hello',
                      routing_key=severity,
                      body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

################################消费者###########################################
import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.1.129')
)

channel = connection.channel()
channel.exchange_declare(
    exchange='hello',
    type='topic'    #topic类型
)

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

binding_key = ['info.#']

for items in binding_key:
    channel.queue_bind(
        exchange='hello',
        queue=queue_name,
        routing_key=items
    )
print('Wating for messages. To exit press CTRL+C')

def callback(ch, method, properties, body):
    '''
    回调方法,当我们获取到消息的时候，Pika库就会调用此回调函数。这个回调函数会将接收到的消息内容输出到屏幕上。
    :param ch:  虚拟通道
    :param method: 方法
    :param properties: 消息属性
    :param body:  消息主体
    :return:
    '''
    print('Received: %r' %body)
    ch.basic_ack(delivery_tag=method.delivery_tag)


while True:
    channel.basic_consume(    #需要告诉RabbitMQ这个回调函数将会从名为"hello"的队列中接收消息：
        callback,
        queue=queue_name,
        no_ack=False
    )
    channel.start_consuming()