python学习之-- RabbitMQ 消息队列
记录：异步网络框架：twisted学习参考：www.cnblogs.com/alex3714/articles/5248247.html
RabbitMQ 模块 《消息队列》
先说明：python的队列
1：线程 queue 只在同一进程内的线程间交互数据
2：进程 queue 只在同一父进程及子进程间交互数据
只应用于python，无法和其他语言程序通信
消息队列有如下几种：（Rabbitmq,ZeroMq,ActiveMq）
功能：可以实现，不同程序间的数据交互
安装：
1：下载安装erlang，因为rabbitmq是erlang开发的
yum install ncurses-devel gcc 安装关联包，如果安装了就跳过
解压 otp_src 包：configure;make &
2：下载rabbitmq并进行安装
2：pycharm安装pika 模块
消息分发轮询：当同时出现多个消费者进入监听模式时，生产者发送的消息会轮询的被多个消费者接收到。
消息持久化：当mq服务器关闭服务然后重启后，发现消息队列清空了，是因为消息存储在内存中，保持消息持久化需要对生产者和消费者声明队列持久化参数，并对生产者的消息进行持久化
参数如：
durable=True：进行队列持久化
delivery_mode=2： 进行生产者的消息持久化
举例代码如：
生产者
# rabbitmq发送端
import pika
# 连接 mq服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
# 声明一个管道
channel = connection.channel()
# 在管道中声明一个队列
channel.queue_declare(queue='hello1',durable=True)# durable 进行队列持久化
# 在指定队列中发送一条消息 生产者
channel.basic_publish(exchange='',
routing_key='hello1',# 消息队列名
body='Hello World',# 消息内容
properties = pika.BasicProperties(
delivery_mode=2,# 进行队列的消息持久化
))
# 发送完毕 关闭连接
print('xx send "Hello World"')
connection.close()
View Code 消费者
# rabbitmq 接收端
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello1',durable=True)# 进行队列持久化
# 打印并处理接收到的消息
def callback(ch,method,properties,body):# 回调函数
print('>>',ch,method,properties)
print('xx Received %r' % body)
ch.basic_ack(delivery_tag=method.delivery_tag)# 手工与服务器端确认消息接收模式
# 可选：指定消费者处理队列消息的数量
channel.basic_qos(prefetch_count=1)# 实现消费者负载均衡，表示同时只能处理1条消息，待处理完成后在接收下一条消息
# 消费者，接收消息如果有就调用callback函数处理消息
channel.basic_consume(callback,queue='hello1',
# no_ack=True# 这个参数是实现自动和服务器端确认，开启这个就不用开启手工确认模式。
)
# 完成 打印
print('xx Waiting for messages. To exit press CTRL+C')
# 启动管道，并处于监听模式
channel.start_consuming()
View Code 广播模式
特点：当发布方发布消息后，如果接收方没有启动服务，则消息就丢失不会接收到。
exchange：指定具有类型，决定哪些queue符合条件，可以接受消息
类型1：fanout：所有绑定到此exchange的queue都可以接受消息,(如果消费者没有启动接收，那么发送者发送的消息只有启动了的消费者能收到，未启动的就不会接收到，广播模式)，这个就是订阅发布
类型2：direct：通过routingkey和exchange决定的那个唯一的queue可以接收消息
类型3：topic：所有符合routingkey（这里可以是一个表达式）的routingkey所bind的queue可以接收消息
表达式符号说明： # 代表一个或多个字符，*代表任何字符
如：#.a 匹配a.a,aa.a,aaa.a等 *.a匹配 a.a,b.a,c.a等
注：随用routingkey为#，exchange type 为topic的时候相当于使用fanout
headers：通过headers来决定把消息发给哪些queue， 用处较少
fanout模式举例：
生产者（就是发消息的）：
import pika
# 连接 mq服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
# 声明一个管道
channel = connection.channel()
# 生产者声明为广播模式的fanout类型，消息发送到logs中。消费者的队列从logs中接收消息
channel.exchange_declare(exchange='logs',type='fanout')# 类型为fanout的exchange广播模式
# message = ''.join(sys.argv) or 'info:Hello World'
message = 'info:Hello World'#广播的消息
# basic_publish为发布消息方法
channel.basic_publish(exchange='logs', # 消息发送至logs
routing_key='',# 广播模式不需要队列名
body=message)# 广播的消息内容
print('xx send "Hello World"')
connection.close()
View Code 消费者（就是接收消息的）：
import pika
# 当为广播模式的fanout时候，原理说明：生产者只会将消息发送到exchange绑定的logs里，无法将消息直接发送给消费者，
# 而消费者必须通过队列来接收消息，所以消费者每次接收消息时，都需要随机生成一个队列名，然后与生产者的logs进行绑定，
# 这时就可以从logs接收到消息，最后消费者在从队列里提取消息。
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 消费者也需要声明广播类型及消息转发器logs
channel.exchange_declare(exchange='logs',type='fanout')
result = channel.queue_declare(exclusive=True)# exclusive（唯一的），不指定queue名字，rabbit会随机分配一个名字，
#exclusize=True会在使用此queue的消费者断开后，自动将queue删除
# 这里result为随机生成的queue 对象
queue_name = result.method.queue# 消费者用于接收消息 随机生成的队列名
channel.queue_bind(exchange='logs',# 将转发器logs和本地的随机生成的队列名进行绑定
queue=queue_name)
# 打印并处理接收到的消息
def callback(ch,method,properties,body):# 回调函数
print('>>',ch,method,properties)
print('xx Received %r' % body)
ch.basic_ack(delivery_tag=method.delivery_tag)# 手工确认回复并断开
# basic_consume 方法实现从随机队列中接收消息
channel.basic_consume(callback,# 调用回调函数，进行消息的接收处理
queue=queue_name,# 接收消息的队列名，广播模式下生产者广播消息不用使用队列，但消费者必须从队列来接收消息，
# no_ack=True# 这个参数是实现自动和服务器端确认，开启这个就不用开启手工确认模式。
)
print('
[*] Waiting for logs.To exit press CTRL+C')
channel.start_consuming()
View Code direct模式（过滤）举例：
# 注意：通用功能性参数解释同上
生产者（消息广播者）：
import pika,sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',# 消息转发器
type='direct')# 广播类型
serverity = sys.argv if len(sys.argv) > 1 else 'info'# 日志级别
message = ''.join(sys.argv) or 'Hello World'# 消息内容
# 将消息通过转发器发送到级别为：serverity
# 注意：就这里和fanout 不同
channel.basic_publish(exchange='direct_logs',routing_key=serverity,body=message)# 表示消息发送到serverity级别内
print('xx send %r:%r' % (serverity,message))
connection.close()
View Code 消费者（消息接收者）：
import pika,sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
serverities = sys.argv# 获取脚本所有参数，就是日志级别列表
if not serverities:
sys.stderr.write('Usage:%s \n' % sys.argv)
sys.exit(1)
# 循环日志级别列表，并绑定到消息队列，指定哪些消息发送到哪个日志级别
for serverity in serverities:
# 循环每个日志级别并与随机队列名和转发器的绑定
channel.queue_bind(exchange='direct_logs',queue=queue_name,routing_key=serverity)
def callback(ch,method,properties,body):
# 消息处理函数
print(' %r:%r' %(method.routing_key,body))
# 接收消息方法
channel.basic_consume(callback,queue=queue_name,no_ack=True)
print('
[*] Waiting for logs,To exit press CTRL+C')
channel.start_consuming()
View Code topic模式（细致的消息过滤模式）举例：
脚本参数说明：带 # 号表示收取所有消息， kern.* ： 收取kern的， kern.* *.critical：表示同时可以收取这2个标识的消息
生产者：
import pika,sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
type='topic')# 广播类型
routing_key = sys.argv if len(sys.argv) > 1 else 'anonymous.info'
message = ''.join(sys.argv) or 'Hello World'
# 广播消息
channel.basic_publish(exchange='topic_logs',# 广播消息转发器
routing_key=routing_key,# 日志级别
body=message)
print(' sent %r:%r' % (routing_key,message))
connection.close()
View Code 消费者
import pika,sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',type='topic')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
bind_key = sys.argv
if not bind_key:
sys.stderr.write('Usage: %s ...\n' % sys.argv)
sys.exit(1)
for binding_key in bind_key:# 循环命令参数，进行对队列名和转发器的绑定
channel.queue_bind(exchange='topic_logs',queue=queue_name,routing_key=binding_key)
print('
[*] Waiting for logs. to exit press CTRL+C')
def callback(ch,method,properties,body):
print(' %r:%r' % (method.routing_key,body))
# 接收消息
channel.basic_consume(callback,queue=queue_name,no_ack=True)
channel.start_consuming()
View Code MQrpc：一端给另一端发送消息后，另一端可以回复消息，典型应用是snmp
举例：
客户端：
import pika
import uuid
# 实现原理：同时建立发消息的队列名和用于返回消息的随机队列名，且每次发送消息都要生成一个唯一的随机uuid值，功能是用于
#做消息返回时通过这个值对比确认是否为所发送消息的返回值，启动客户端将消息发送到服务器后，消息包包含：消息内容，随机生成
#的消息队列名，唯一UUID，待服务器端接收到消息后，处理完成将处理结果和uuid通过随机的队列名返回给客户端，这时客户端从自己
#生成的随机队列内可以接收到服务器发来的消息。
class FibonacciRpcClient(object):
def __init__(self):
self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
self.channel = self.connection.channel()
result = self.channel.queue_declare(exclusive=True)# 生成一个随机的队列内存地址，功能用于消息的返回使用
self.callback_queue = result.method.queue# 随机生成的队列名，用于发送给服务器端，服务器端将结果通过此队列返回给客户端
# 接收消息
self.channel.basic_consume(self.on_response,# 调用on_response函数
no_ack=True,
queue=self.callback_queue# 指定队列为本机随机生成的队列
)
def on_response(self,ch,method,props,body):
# 接收消息并处理
if self.corr_id == props.correlation_id:# 判断本地生成的ID是否和服务器端返回的ID相同，如相同则确认为本次的返回结果
self.response = body# body为本次发送所返回的数据
def call(self,n):# 实现消息的发送和接收
self.response = None# 首先声明接收到消息为空
self.corr_id = str(uuid.uuid4())# 生成一个随机的id值
self.channel.basic_publish(exchange='',# 发送消息
routing_key='rpc_queue',# 消息发送到服务器端的队列名
properties=pika.BasicProperties(
# 以下2个参数和发送的消息一起发送到服务器端
reply_to=self.callback_queue,# 告诉服务器端返回消息的队列名
correlation_id=self.corr_id,# 本次发送消息随机生成的ID值
),
body=str(n))# 发送到服务器的消息内容
while self.response is None:# 接收服务器端返回的消息，为空进入无限循环
self.connection.process_data_events()# 非阻塞版的start_consuming()，有无消息都会返回
return int(self.response)# 返回服务器端返回的数据
fibonacci_rpc = FibonacciRpcClient()# 实例化
print(' Requesting fib')
response = fibonacci_rpc.call(30)# 调用call执行
print('[.] Got %r' % response)
View Code 服务器端：
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()# 声明一个通道
channel.queue_declare(queue='rpc_queue')# 声明一个队列，为客户端发送消息指定的队列名
def fib(n):# 收到客户端消息进行处理的函数，并返回处理结果
'''功能：实现斐波那切'''
if n == 0:
return 0
elif n == 1:
return 1
else:
return fib(n-1) + fib(n-2)
def on_request(ch,method,props,body):
n = int(body)# 接收到的消息
response = fib(n)# 调用函数处理接收到的消息并返回处理结果
ch.basic_publish(exchange='',# 进行客户端的结果返回
routing_key=props.reply_to,# 这里的队列名就是客户端发来的随机生成的队列名
properties=pika.BasicProperties(correlation_id=props.correlation_id),# 返回的属性信息（就是客户端发来的ID值）
body=str(response))# 返回给客户端的fib函数执行结果的数据
ch.basic_ack(delivery_tag=method.delivery_tag)# 手动结束本次队列连接
# channel.basic_qos(prefetch_count=1)# 增加这个则一次只能处理对立中一条信息
channel.basic_consume(on_request,queue='rpc_queue')# 接收客户端发来的消息
print(' Awaiting RPC requests')
channel.start_consuming()# 启动程序，进入监听模式
View Code
页:
[1]