python第六十天－－－－－RabbitMQ

RabbitMQ消息队列：默认为消息轮循模式，按client端启动是顺序接收

server端

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()#管道

#声明queue
#channel.queue_declare(queue='hello')#队列名 hello
channel.queue_declare(queue='hello',durable=True)#队列名 hello,持久化队列

for i in range(10):

    channel.basic_publish(exchange='',
                          routing_key='hello',
                          body='Hello World!%s'%i,
                          properties=pika.BasicProperties(delivery_mode=2))
    print(" [x] Sent 'Hello World!'",i)
connection.close()

client端

import pika,time

connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()
#channel.queue_declare(queue='hello')
channel.queue_declare(queue='hello',durable=True)#队列名 hello,持久化队列


def callback(ch, method, properties, body):#回调函数
    print('接收消息中…………')
    time.sleep(1)
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)#同时只处理一个消息
channel.basic_consume(callback,#接收到消息调用回调函数 callback
                      queue='hello',
                      #no_ack=True
                       )

print(' [*] 接收消息中. To exit press CTRL+C')
channel.start_consuming()#启动消息接收

消息持久化

server端

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()#管道

#声明queue
channel.queue_declare(queue='hello2',durable=True)#队列名 hello
#channel.queue_declare(queue='hello2',durable=True)#队列名 hello,持久化队列


channel.basic_publish(exchange='',
                        routing_key='hello2',
                        body='Hello World!%s---->')
print(" [x] Sent 'Hello World!'")
connection.close()

client端

import pika,time

connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello2')#服务端与客户端的设置需一致，不然会报错
#channel.queue_declare(queue='hello2',durable=True)#队列名 hello,持久化队列


def callback(ch, method, properties, body):#回调函数
    print('接收消息中…………')
    time.sleep(5)
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)#同时只处理一个消息
channel.basic_consume(callback,#接收到消息调用回调函数 callback
                      queue='hello2',
                      #no_ack=True
                       )

print(' [*] 接收消息中. To exit press CTRL+C')
channel.start_consuming()#启动消息接收

fanout广播模式：实时发送，发送时，如果client端没启动将无法收到消息

server端

import pika,sys,time
connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()#管道

#声明queue 广播模式不用声明队列
#channel.queue_declare(queue='hello')#队列名 hello
#channel.queue_declare(queue='hello',durable=True)#队列名 hello,持久化队列

argv=input('输入消息')
msg=''.join(sys.argv[1:]) or  'info:消息默认发送………'
for i in range(10):
    time.sleep(1)
    channel.basic_publish(exchange='logs',#绑定频道
                          #routing_key='hello',
                          routing_key='',
                          body=msg+str(i),
                          #properties=pika.BasicProperties(delivery_mode=2)#持久化 广播不能使用
                           )
    print(msg,i)
#connection.close()

client端

import pika,time

connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()
#channel.queue_declare(queue='hello2')#服务端与客户端的设置需一致，不然会报错
#channel.queue_declare(queue='hello2',durable=True)#队列名 hello,持久化队列
channel.exchange_declare(exchange='logs',#绑定频道
                         type='fanout')#接收类型
reult=channel.queue_declare(exclusive=True)#随机生成唯一的队列名，会在消息接收后自动删除
queuename=reult.method.queue#队列名 自动生成
channel.queue_bind(exchange='logs',#先要绑定频道
                   queue=queuename
                   )


def callback(ch, method, properties, body):#回调函数
    print('接收消息中…………')
    #time.sleep(5)
    print(" [x] Received %r" % body.decode())
    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)#同时只处理一个消息
channel.basic_consume(callback,#接收到消息调用回调函数 callback
                      queue=queuename,
                      #no_ack=True
                       )

print(' [*] 接收消息中. To exit press CTRL+C')

channel.start_consuming()#启动消息接收

direct广播模式：分级别发送消——客户端可以按级别接收

server端

import pika,sys,time
connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()#管道



severity = sys.argv[1] if len(sys.argv) > 1 else 'info'#启动参数 默认无参数为 info 级别
msg=''.join(sys.argv[2:]) or  'info:消息默认发送………'#启动参数 为空，发默认消息
for i in range(10):
    time.sleep(1)
    channel.basic_publish(exchange='direct_logs',#绑定频道
                          routing_key=severity,#默认的消息队列级别
                          body=msg+str(i),
                          #properties=pika.BasicProperties(delivery_mode=2)#持久化 广播不能使用
                           )
    print(msg,severity)
connection.close()

client端

import pika,time,sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',#定义一个接收的频道
                         type='direct')

reult=channel.queue_declare(exclusive=True)#随机生成唯一的队列名，会在消息接收后自动删除
queuename=reult.method.queue#队列名 自动生成


severities = sys.argv[1:]
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='direct_logs',
                       queue=queuename,
                       routing_key=severity)

def callback(ch, method, properties, body):#回调函数
    print('接收消息中…………')
    #time.sleep(5)
    print(" [x] Received %r" % body.decode())
    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)#同时只处理一个消息
channel.basic_consume(callback,#接收到消息调用回调函数 callback
                      queue=queuename,
                      #no_ack=True
                       )

print(' [*] 接收消息中. To exit press CTRL+C')

channel.start_consuming()#启动消息接收

topic细致消息过滤

server端

import pika,sys,time
connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()#管道



#severity = sys.argv[1] if len(sys.argv) > 1 else 'info'#启动参数 默认无参数为 info 级别
routing_key= sys.argv[1] if len(sys.argv) > 1 else 'anorrymous.info'#启动参数 默认无参数为 info 级别
msg=''.join(sys.argv[2:]) or  'info:消息默认发送………'#启动参数 为空，发默认消息
for i in range(10):
    time.sleep(1)
    channel.basic_publish(exchange='direct_logs',#绑定频道
                          #routing_key=severity,#默认的消息队列级别
                          routing_key=routing_key,#默认的消息队列级别
                          body=msg+str(i),
                          #properties=pika.BasicProperties(delivery_mode=2)#持久化 广播不能使用
                           )
    #print(msg,severity)
    print(msg,routing_key)
connection.close()

client端

import pika,time,sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',#定义一个接收的频道
                         type='topic')

reult=channel.queue_declare(exclusive=True)#随机生成唯一的队列名，会在消息接收后自动删除
queuename=reult.method.queue#队列名 自动生成


#severities = sys.argv[1:]
binding_key = sys.argv[1:]
#if not severities:
if not binding_key:
    sys.stderr.write("Usage: %s [info] [warning] [error]
" % sys.argv[0])#启动接收的消息级别
    sys.exit(1)

#for severity in severities:#循环接收各级别的消息
for severity in binding_key:#循环接收各级别的消息
    channel.queue_bind(exchange='direct_logs',
                       queue=queuename,
                       routing_key=severity)

def callback(ch, method, properties, body):#回调函数
    print('接收消息中…………')
    #time.sleep(5)
    print(" [x] Received %r" % body.decode())
    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)#同时只处理一个消息
channel.basic_consume(callback,#接收到消息调用回调函数 callback
                      queue=queuename,
                      #no_ack=True
                       )

print(' [*] 接收消息中. To exit press CTRL+C')

channel.start_consuming()#启动消息接收

RPC模式：结果返回

server端

import pika
import time
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)

    print(" [.] fib(%s)" % n)
    response = fib(n)#调用计算函数

    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,#收消息的队列
                     properties=pika.BasicProperties(correlation_id =props.correlation_id),#返回消息的队列
                     body=str(response))#返回结果数据
    ch.basic_ack(delivery_tag = method.delivery_tag)##确保消息被 客户端接收

channel.basic_qos(prefetch_count=1)#同时只处理一个消息
channel.basic_consume(on_request, queue='rpc_queue')#接收消息，自动调用回调函数

print(" [x] Awaiting RPC requests")
channel.start_consuming()#开始接收

client端

import pika
import uuid

class FibonacciRpcClient(object):
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(
                host='localhost'))#生成连接的服务端 ip

        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,#回调函数
                                   no_ack=True,#不用服务端确认本条消息
                                   queue=self.callback_queue)#随机生的队列

    def on_response(self, ch, method, props, body):#回调函数
        if self.corr_id == props.correlation_id:#判断服务端返回的队列名是否与当前所生成的队列名一致
            self.response = body#  将服务端的结果赋于返回来的结果变量

    def call(self, n):#发送消息的函数
        self.response = None#初始返回结果为空
        self.corr_id = str(uuid.uuid4())#生成一个服务端返回消息的队列名
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',#确认为rpc模式 实时发送
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,#发送的管道队列名
                                         correlation_id = self.corr_id,#发送给服务端，用于返回消息的队列名
                                         ),
                                   body=str(n))#发送的内容数据
        while self.response is None:
            self.connection.process_data_events()#非阻塞模式接收消息
            print('没有返回消息……')
        return int(self.response)#返回结果

fibonacci_rpc = FibonacciRpcClient()#生成一个实例

print(" [x] Requesting fib(10)")
response = fibonacci_rpc.call(10)#调用发送消息的函数
print(" [.] Got %r" % response)#打印结果