论坛 › Exchange › python网络篇【第十二篇】RabbitMQ

haixin3036 发表于 2017-7-2 15:39:02

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(" 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(" 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(" 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 \n" % sys.argv)
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(" %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 \n" % sys.argv)
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(" %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(" 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()
订阅

查看完整版本: python网络篇【第十二篇】RabbitMQ