python2.0_s12_day10_rabbitMQ使用介绍
RabbitMQ
RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。
MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。
RabbitMQ安装
Linux 安装配置epel源 $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm 安装erlang,因为RabiitMQ用erlang语言写的 $ yum -y install erlang 安装RabbitMQ $ yum -y install rabbitmq-server 注意：service rabbitmq-server start/stop
MAC 安装 http://www.rabbitmq.com/install-standalone-mac.html
安装API
pip install pika #pika是官方提供的,当然还有其他的 or easy_install pika or 源码 https://pypi.python.org/pypi/pika
一、实现最简单的队列通信
send端
1 #!/usr/bin/evn python3.5 2 #__author__:"ted.zhou" 3 ''' 4 zibbitMQ最简单的队列通信代码范例 5 ''' 6 import pika 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) # 连接一个rabbitMQ,返回连接成功后的实例 9 channel = connection.channel() # 创建一个管道,用于传输各种队列.--连接成功后,还不能直接使用,需要在这个连接的实例中创建一个管道. 10 11 # 声明一个queue 12 # channel.queue_declare(queue='hello') # 在这个管道里声明一个队列 ,队列的名称为"hello" 13 ''' 14 使用pika连接并创建队列需要三步 15 1.使用pika.BlockingConnection() 创建一个连接 16 2.创建一个管道 17 3.声明一个队列 18 ''' 19 20 # 紧接着就可以通过这个管道发送内容了,在发送时,必须有三个参数 21 # exchege = '' 这个在发布订阅模式时,会用到,具体高级用法会提到,这里默认给'',这样它内部还是会调用一个默认类型. 22 # routing_key = 'hello',这里的routing_key 是选择通过哪个队列发送 23 # body = 'Hello World!' 要发送的内容 24 channel.basic_publish(exchange='', 25 routing_key='hello', # 接收端不是这个参数,而是queue 26 body='Hello World!') 27 28 print(" [x] Sent 'Hello World!'") # 生产者端打印发送信息,表示代码已经执行到这里 29 connection.close() # 关闭这个连接
receive端
1 #!/usr/bin/env python3.5 2 __author__ = "ted.zhou" 3 ''' 4 python使用zabbitMQ实现最简单的队列通信之接收端代码范例 5 ''' 6 7 import pika 8 9 # 使用pika模块,连接到指定的rabbitMQ服务器 10 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 11 12 # 连接创建成功后,实例化一个管道 13 channel = connection.channel() 14 15 # 然后在管道中声明一个队列,表示我这个管道里可以跑 'hello'这个队列, 16 # 我们在发送端声明了一个'hello' 的queue,这里为什么还要声明一次 ,因为当接收端先启动的时候,此时不声明,下面代码在接收时会报错. 17 # 当然发送端如果先启动了,这里声明也不会报错. 18 19 20 channel.queue_declare(queue='hello') 21 22 # 紧接着我们就要进行接收队列里的消息,但是接收之前我们要知道这个消息我们收来做哪些操作呢,只接过来没啥意义. 23 # 所以定义一个callback函数 24 25 # 这里注意,接收端定义的callback函数,一定要带三个参数 26 # 1.ch 2.method 3.properties 4.其后才是信息主题body 27 # 前面3个参数是做什么的,暂时用不到,后面高级的用法会举例 28 def callback(ch,method,properties,body): 29 print("[x] Received %r" %body) 30 31 # 紧接着定义接收,定义完接收并不是直接就接收了,这个和发送端的basic_publish()方法不太一样,basic_publish()是直接就发送了,而接收basic_consume()方法定义后,还需要调用一个start方法 32 # 定义管道的接收方法. 33 # 参数介绍: queue 指定 接收的队列名称 , no_ack=True 是定义此接收方法是否要确认执行完成,如果为True, 34 # 说明不需要验证执行状态,也就是说当一个callback需要处理6分钟,当5分钟时程序卡死了,此消息也就没了,如果为False,5分钟卡死后,消息在队列中依然存在 35 channel.basic_consume(callback, 36 queue='hello', # 发送端不是这个参数,而是routing_key 37 no_ack=True) 38 print(' [*] Waiting for messages. To exit press CTRL+C') 39 channel.start_consuming() # 开启接收,没有就阻塞
晋级:
二、队列持久化&消息持久化:
我们上面的例子,在发送端管道cannel中声明了'hello'队列.
为了避免当接收端先启动的情况下,因为发送端还未运行程序导致rabbitMQ服务中没有'hello'队列,导致接收端程序报错,所以在接收端中的管道也声明了'hello'队列
无论是发送端还是接收端在管道cannel中声明了'hello'队列,在rabbitMQ服务器中,你都可以通过命令查看此队列的信息:
MacBook-Pro:~ tedzhou$ sudo rabbitmqctl list_queues
Listing queues ...
hello 0
那么问题来了,当发送端发送了很多信息在'hello'队列中,接收端还没启动呢,这时候所有的信息都存在hello队列,如下这种情况:
MacBook-Pro:~ tedzhou$ sudo rabbitmqctl list_queues
Listing queues ...
hello 7
如果此时rabbitMQ服务器挂了,或者重启了,会有两个问题:1.这个'hello'队列还存在吗? 2.'hello'队列中的信息还存在吗?
我们做下测试：
停止rabbitMQ服务 MacBook-Pro:~ tedzhou$ sudo rabbitmqctl stop Stopping and halting node 'rabbit@zhoumingdeMacBook-Pro' ... 启动rabbitMQ服务 MacBook-Pro:~ tedzhou$ sudo rabbitmq-server 查看rabbitMQ的队列 MacBook-Pro:~ tedzhou$ sudo rabbitmqctl list_queues Listing queues ... 结果证明了: 1.队列没有了 2.消息更没有了
整成业务中,我们肯定希望这些队列和消息能够保留下来.所以我们要解决两个问题.
1.持久化队列
2.持久化消息
1.队列持久化代码范例
要在声明队列的时候,加上队列持久化参数
channel.queue_declare(queue='hello', durable=True)
2.消息持久化代码范例
要在发送消息的代码部分,加上消息持久化的属性,delivery_mode=2就是说这个消息持久化消息,直到消费掉.(老实说delivery_mode有30多种,常用的就这一种)
channel.basic_publish(exchange='',
routing_key="task_queue",
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))
发送端要在声明队列和发送消息中更改代码
1 #!/usr/bin/evn python3.5 2 #__author__:"ted.zhou" 3 ''' 4 zibbitMQ最简单的队列通信代码范例 5 ''' 6 import pika 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) # 连接一个rabbitMQ,返回连接成功后的实例 9 channel = connection.channel() # 创建一个管道,用于传输各种队列.--连接成功后,还不能直接使用,需要在这个连接的实例中创建一个管道. 10 11 # 声明一个queue 12 # channel.queue_declare(queue='hello') # 在这个管道里声明一个队列 ,队列的名称为"hello" 13 channel.queue_declare(queue='hello',durable=True) # durable=True 设置此队列持久化属性为True 14 ''' 15 使用pika连接并创建队列需要三步 16 1.使用pika.BlockingConnection() 创建一个连接 17 2.创建一个管道 18 3.声明一个队列 19 ''' 20 21 # 紧接着就可以通过这个管道发送内容了,在发送时,必须有三个参数 22 # exchege = '' 这个在发布订阅模式时,会用到,具体高级用法会提到,这里默认给'',这样它内部还是会调用一个默认类型. 23 # routing_key = 'hello',这里的routing_key 是选择通过哪个队列发送 24 # body = 'Hello World!' 要发送的内容 25 channel.basic_publish(exchange='', 26 routing_key='hello', # 接收端不是这个参数,而是queue 27 body='Hello World!', 28 properties=pika.BasicProperties( # 消息持久化加入的参数 29 delivery_mode = 2,) 30 ) 31 32 print(" [x] Sent 'Hello World!'") # 生产者端打印发送信息,表示代码已经执行到这里 33 connection.close() # 关闭这个连接
接收端1.需要在声明队列中设置持久化属性,2.它要在callback中获得接收到的数据de
1 #!/usr/bin/env python3.5 2 __author__ = "ted.zhou" 3 ''' 4 python使用zabbitMQ实现最简单的队列通信之接收端代码范例 5 ''' 6 7 import pika 8 9 # 使用pika模块,连接到指定的rabbitMQ服务器 10 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 11 12 # 连接创建成功后,实例化一个管道 13 channel = connection.channel() 14 15 # 然后在管道中声明一个队列,表示我这个管道里可以跑 'hello'这个队列, 16 # 我们在发送端声明了一个'hello' 的queue,这里为什么还要声明一次 ,因为当接收端先启动的时候,此时不声明,下面代码在接收时会报错. 17 # 当然发送端如果先启动了,这里声明也不会报错. 18 19 20 #channel.queue_declare(queue='hello') 21 channel.queue_declare(queue='hello',durable=True) #durable=True 设置此队列持久化属性为True 22 23 24 # 紧接着我们就要进行接收队列里的消息,但是接收之前我们要知道这个消息我们收来做哪些操作呢,只接过来没啥意义. 25 # 所以定义一个callback函数 26 27 # 这里注意,接收端定义的callback函数,一定要带三个参数 28 # 1.ch 2.method 3.properties 4.其后才是信息主题body 29 # 前面3个参数是做什么的,暂时用不到,后面高级的用法会举例 30 def callback(ch,method,properties,body): 31 print("[x] Received %r" %body) 32 time.sleep(body.count(b'.')) 33 print(" [x] Done") 34 ch.basic_ack(delivery_tag = method.delivery_tag) # 获得delivery_tag,具体啥一起,老师没说,就说咱加上! 35 36 # 紧接着定义接收,定义完接收并不是直接就接收了,这个和发送端的basic_publish()方法不太一样,basic_publish()是直接就发送了,而接收basic_consume()方法定义后,还需要调用一个start方法 37 # 定义管道的接收方法. 38 # 参数介绍: queue 指定 接收的队列名称 , no_ack=True 是定义此接收方法是否要确认执行完成,如果为True, 39 # 说明不需要验证执行状态,也就是说当一个callback需要处理6分钟,当5分钟时程序卡死了,此消息也就没了,如果为False,5分钟卡死后,消息在队列中依然存在 40 channel.basic_consume(callback, 41 queue='hello') # 发送端不是这个参数,而是routing_key 42 43 print(' [*] Waiting for messages. To exit press CTRL+C') 44 channel.start_consuming() # 开启接收,没有就阻塞
我们通过查看rabbitMQ里的队列情况,来验证下是否持久化成功.
1 首先只运行发送端程序,运行6遍. 2 查看队列: 3 MacBook-Pro:~ tedzhou$ sudo rabbitmqctl list_queues 4 Listing queues ... 5 hello 6 6 停掉服务: 7 MacBook-Pro:~ tedzhou$ sudo rabbitmqctl stop 8 Stopping and halting node 'rabbit@zhoumingdeMacBook-Pro' ... 9 开启服务: 10 MacBook-Pro:~ tedzhou$sudo rabbitmq-server & 11 再次查看队列: 12 MacBook-Pro:~ tedzhou$ sudo rabbitmqctl list_queues 13 Listing queues ... 14 hello 6
验证结果: 持久化 队列&消息成功.
用法晋级2
三.Work Queues
在这种模式下，RabbitMQ会默认把p发的消息依次分发给各个消费者(c),跟负载均衡差不多
消息生产者代码
1 import pika 2 3 connection = pika.BlockingConnection(pika.ConnectionParameters( 4 'localhost')) 5 channel = connection.channel() 6 7 #声明queue 8 channel.queue_declare(queue='task_queue') 9 10 #n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange. 11 import sys 12 13 message = ' '.join(sys.argv[1:]) or "Hello World!" 14 channel.basic_publish(exchange='', 15 routing_key='task_queue', 16 body=message, 17 properties=pika.BasicProperties( 18 delivery_mode = 2, # make message persistent 19 )) 20 print(" [x] Sent %r" % message) 21 connection.close()
消费者代码:
1 import pika,time 2 3 connection = pika.BlockingConnection(pika.ConnectionParameters( 4 'localhost')) 5 channel = connection.channel() 6 7 8 9 def callback(ch, method, properties, body): 10 print(" [x] Received %r" % body) 11 time.sleep(body.count(b'.')) 12 print(" [x] Done") 13 ch.basic_ack(delivery_tag = method.delivery_tag) 14 15 16 channel.basic_consume(callback, 17 queue='task_queue', 18 ) 19 20 print(' [*] Waiting for messages. To exit press CTRL+C') 21 channel.start_consuming()
当你多次运行一个生产者的代码,而运行3个消费者的代码,你会发现消息会轮询3个消费者程序,也就是消费者会依次接收到代码,这个就像简单的负载均衡.
那么问题来了,加入运行消费者程序的3台机器的配置不一样,好的1台,消费一条消息需要1分钟, 性能差的机器要10分钟,那么前面说到的负载均衡就会导致,差的严重影响效率.
我们在LVS这类负载均衡是可以设置权重,同样消费者在接收消息时也可以设置相应的功能,但不是权重,它比权重更人性化,它可以保证一个消费者程序,同时只能保证1个信息在消费,当然也可以设置同一时刻保证在消费2个信息
具体实现代码如下:
生产者代码不变:
1 #!/usr/bin/env python 2 import pika 3 import sys 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host='localhost')) 7 channel = connection.channel() 8 9 channel.queue_declare(queue='task_queue', durable=True) 10 11 message = ' '.join(sys.argv[1:]) or "Hello World!" 12 channel.basic_publish(exchange='', 13 routing_key='task_queue', 14 body=message, 15 properties=pika.BasicProperties( 16 delivery_mode = 2, # make message persistent 17 )) 18 print(" [x] Sent %r" % message) 19 connection.close()
消费者代码加入channel.basic_qos(prefetch_count=1),代码如下:
1 #!/usr/bin/env python 2 import pika 3 import time 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host='localhost')) 7 channel = connection.channel() 8 9 channel.queue_declare(queue='task_queue', durable=True) 10 print(' [*] Waiting for messages. To exit press CTRL+C') 11 12 def callback(ch, method, properties, body): 13 print(" [x] Received %r" % body) 14 time.sleep(body.count(b'.')) 15 print(" [x] Done") 16 ch.basic_ack(delivery_tag = method.delivery_tag) 17 18 channel.basic_qos(prefetch_count=1) ＃表示同意时刻保证客户端程序只处理一个消息 19 channel.basic_consume(callback, 20 queue='task_queue') 21 22 channel.start_consuming()
rabbitMQ高级用法
四、PublishSubscribe(消息发布订阅)
之前的例子都基本都是1对1的消息发送和接收，即消息只能发送到指定的queue里，但有些时候你想让你的消息被所有的Queue收到，类似广播的效果，这时候就要用到exchange了，
Exchange在定义的时候是有类型的，以决定到底是哪些Queue符合条件，可以接收消息
fanout: 所有bind到此exchange的queue都可以接收消息
direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息
topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息，不是通过queue名来过滤，可以监控关键字，具体用法
表达式符号说明：#代表一个或多个字符，*代表任何字符
例：#.a会匹配a.a，aa.a，aaa.a等
*.a会匹配a.a，b.a，c.a等
注：使用RoutingKey为#，Exchange Type为topic的时候相当于使用fanout
headers: 通过headers 来决定把消息发给哪些queue ＃这个用的非常少，官网上暂时没有例子
exchange 的 fanout 类型举例（广播给所有bind的queue）
消息publisher
1 #!/usr/bin/env python3.5 2 #__author__:'ted.zhou' 3 ''' 4 使用pika模块中的exchage的fanout类型,广播发送消息,所有绑定了类型为fanout的exchage的queue都会收到广播信息,且发送端不要在指定发送路由routing_key 5 ''' 6 import pika 7 import sys 8 9 connection = pika.BlockingConnection(pika.ConnectionParameters( 10 host='localhost')) 11 channel = connection.channel() 12 13 channel.exchange_declare(exchange='logs',type='fanout') 14 # 这里声明了exchange,但是没有声明queue,因为exchange的fanout类型,默认是广播给所有绑定了此exchange的队列. 15 # 广播发送端,就没必要在发送的时候指定我是通过哪个queue发送了.因为默认绑定这个exchange的queue都会路由.所以下面发送的时候routing=key=''为空 16 17 18 message = ' '.join(sys.argv[1:]) or "info: Hello World!" # 如果执行此程序时后面没有其他参数,则默认发送内容为"info: Hello World!" 19 20 channel.basic_publish(exchange='logs', 21 routing_key='', 22 body=message) 23 print(" [x] Sent %r" % message) 24 connection.close()
消息subscriber
1 #_*_coding:utf-8_*_ 2 __author__ = 'ted.zhou' 3 import pika 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) 6 channel = connection.channel() 7 8 channel.exchange_declare(exchange='logs',type='fanout') # 这里也是为了避免客户端先启动时无法找到名为logs的exchage,才在客户端进行声明的 9 10 result = channel.queue_declare(exclusive=True) #不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除 11 queue_name = result.method.queue 12 13 channel.queue_bind(exchange='logs',queue=queue_name) # 将此exchage和queue绑定起来 14 15 print(' [*] Waiting for logs. To exit press CTRL+C') 16 17 def callback(ch, method, properties, body): 18 print(" [x] %r" % body) 19 20 channel.basic_consume(callback, 21 queue=queue_name, 22 no_ack=True) 23 24 channel.start_consuming()
有选择的接收消息(exchange type=direct)
RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。
publisher
1 #!/usr/bin/evn python3.5 2 #__author__:"ted.zhou" 3 ''' 4 exchange类型为direct,发送广播消息,只有指定的队列收到该消息 5 ''' 6 import pika 7 import sys 8 9 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) 10 channel = connection.channel() 11 12 channel.exchange_declare(exchange='direct_logs',type='direct') 13 14 severity = sys.argv[1] if len(sys.argv) > 1 else 'info' 15 message = ' '.join(sys.argv[2:]) or 'Hello World!' 16 channel.basic_publish(exchange='direct_logs', 17 routing_key=severity, # 发送内容的关键字,千万不要理解成队列名,这个关键字在exchange=''时,表示队列名.而当exchange不是空时,为广播的关键字 18 body=message) 19 print(" [x] Sent %r:%r" % (severity, message)) 20 connection.close()
subscriber
1 #!/usr/bin/env python3.5 2 #__author__:"ted.zhou" 3 ''' 4 direct类型的exchage接收指定类型的内容 5 ''' 6 import pika 7 import sys 8 9 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) 10 channel = connection.channel() 11 12 channel.exchange_declare(exchange='direct_logs',type='direct') 13 14 result = channel.queue_declare(exclusive=True) #不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除 15 queue_name = result.method.queue # 获取一个随机名称 16 17 severities = sys.argv[1:] # 执行此程序时,参数指接收到的内容的类别,记住不是队列名 18 if not severities: # 如果没指定参数,程序退出 19 sys.stderr.write("Usage: %s [info] [warning] [error] " % sys.argv[0]) 20 sys.exit(1) 21 22 for severity in severities: # 循环绑定exchange,队列,内容类别(内容关键字) 23 channel.queue_bind(exchange='direct_logs', 24 queue=queue_name, 25 routing_key=severity) 26 27 print(' [*] Waiting for logs. To exit press CTRL+C') # 28 29 def callback(ch, method, properties, body): # 定义callback 30 print(" [x] %r:%r" % (method.routing_key, body)) 31 32 channel.basic_consume(callback, 33 queue=queue_name, 34 no_ack=True) 35 36 channel.start_consuming() # 开启接收状态
对direct类型代码的执行测试
更细致的消息过滤(exchange type=topic)
其实direct 和 topic 差不多，都是可以在运行客户端时指定客户端可以获得的消息内容关键字。
区别在于，topic支持类似正则的模式，比如我想获得mysql.error 和apache.*（所有apache.info,apache.error,apache.warnning等）
所以 我觉得如果你有需要根据关键字来接让不同的客户端接收不同的消息，直接用topic类型就好
publisher端代码如下：
1 #!/usr/bin/env python3.5 2 #__author__="ted.zhou" 3 ''' 4 使用exchange的topic类型,支持客户端接收时更细致的过滤 5 ''' 6 import pika 7 import sys 8 9 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) 10 channel = connection.channel() 11 12 channel.exchange_declare(exchange='topic_logs',type='topic') 13 14 routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info' 15 message = ' '.join(sys.argv[2:]) or 'Hello World!' 16 channel.basic_publish(exchange='topic_logs', 17 routing_key=routing_key, 18 body=message) 19 print(" [x] Sent %r:%r" % (routing_key, message)) 20 connection.close()
subscriber代码如下
1 #!/usr/bin/env python3.5 2 __author__='ted.zhou' 3 ''' 4 使用exchange的topic类型,更精细的过滤消息内容 5 ''' 6 import pika 7 import sys 8 9 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) 10 channel = connection.channel() 11 12 channel.exchange_declare(exchange='topic_logs',type='topic') 13 14 result = channel.queue_declare(exclusive=True) 15 queue_name = result.method.queue # 随机产生一个queue名称 16 17 binding_keys = sys.argv[1:] # 获得执行程序的参数 18 if not binding_keys: # 判断是否有参数 19 sys.stderr.write("Usage: %s [binding_key]... " % sys.argv[0]) 20 sys.exit(1) 21 22 for binding_key in binding_keys: #循环绑定 exchange,queue,routing_key 23 channel.queue_bind(exchange='topic_logs', 24 queue=queue_name, 25 routing_key=binding_key) 26 27 print(' [*] Waiting for logs. To exit press CTRL+C') 28 29 def callback(ch, method, properties, body): # 定义callback函数 30 print(" [x] %r:%r" % (method.routing_key, body)) 31 32 channel.basic_consume(callback, 33 queue=queue_name, 34 no_ack=True) 35 36 channel.start_consuming()
对topic类型代码的执行测试
Remote procedure call (RPC)
RPC是远程方法调用,通过客户端发给服务端一个命令,服务端执行命令,把执行结果返回给客户端
这个RPC和刚才的rabbitMQ的几个例子不一样,之前几个例子生产者只把消息发送给消费者,RPC不断要把消息发给你,还要把你的结果反回来
那么问题1来了,负责发送的一端又负责接收,代码上实现接收不难,但是一个管道队列怎样保证发送方发送后,在接收的时候是远程服务器返回来的消息.答案是发送的队列和接收的队列不一致,这样就可以保证消息的准确性.
那么怎样实现 发送命令的队列 和接收返回消息的队列不一致呢.就是发送命令的一端,在发送远程命令的同时,把它将要用来接收返回信息的队列名称也告诉远端服务器.这样就实现了.
不要着急,pika中或者说rabbitMQ中发送信息时可以设置信息属性返回队列的名称.
那么问题2来了,如果你向同一台服务器发送了多个命令,也就是说客户端执行一次程序,发送了多个命令,那么远端服务器收到多次命令后,也进行操作并返回多次的执行结果,那么哪次命令对应哪次结果呢.
这时候我们想到,不仅要在发送信息时设置接收返回信息的队列名称的属性.另外还要给这个发送的信息一个编号.当接收到返回信息时,让远程服务器在发送回来,call端在进行验证.
放心,kipa或者说rabbitMQ同样也支持给信息设置一个属性ID
下面就看下具体实现代码:
rabbitMQ_RPC_call端代码:
1 #!/usr/bin/env python3.5 2 #__author__:"ted.zhou" 3 ''' 4 rabbitMQ实现RPC 5 ''' 6 import pika 7 import uuid 8 9 class FibonacciRpcClient(object): 10 11 def __init__(self): # 定义初始化方法 12 # 实例化连接 13 self.connection = pika.BlockingConnection(pika.ConnectionParameters( 14 host='localhost')) 15 # 实例化管道 16 self.channel = self.connection.channel() 17 18 # 随机生成一个queue名称 19 result = self.channel.queue_declare(exclusive=True) 20 self.callback_queue = result.method.queue 21 22 # 接收方法 23 self.channel.basic_consume(self.on_response, no_ack=True, 24 queue=self.callback_queue) 25 # 接收完,对信息的处理方法 26 def on_response(self, ch, method, props, body): 27 if self.corr_id == props.correlation_id: # 如果corr_id 等于 客户端通过rabbitMQ 的self.callback_queue队列发来的消息的props.correlation_id相等,说明接收到的信息是发送命令的结果 28 self.response = body # 把结果值付给self.response 属性 29 30 # 31 def call(self, n): 32 self.response = None # 先设置response默认为空 33 self.corr_id = str(uuid.uuid4()) # 用uuid随机生成一个uuid值,用来记录程序执行的唯一ID值,主要作用是保证发送和接收内容的一致性,比如你多次执行这个程序,返回来多个值,那么你从管道里取值时,当然要确定这个值是哪次执行程序的结果 34 self.channel.basic_publish(exchange='', 35 routing_key='rpc_queue', 36 properties=pika.BasicProperties( 37 reply_to = self.callback_queue, # reply_to 参数,接收端能够获取,这里用作服务端返回信息时使用的queue 38 correlation_id = self.corr_id,), # 消息ID,这个是多条命令传输时,命令的返回结果和发送的ID保持一致 39 body=str(n)) # 发送的信息 40 while self.response is None: #这段代码的意思是如果self.response值为空,也就是说RPC_server没有返回值,要不就是self.corr_id值不对,要不就是客户端没返回 41 self.connection.process_data_events() # 不断的去这个queue里接收,这个接收不是阻塞,一直去取.,那么问题来了,如果你取的消息和你发送的self.corr_id值不一致,这个消息怎么处理,是放回去,还是删掉.需要测试 42 return int(self.response) # 如果信息得到反馈,则返回得到的计算结果 43 44 fibonacci_rpc = FibonacciRpcClient() # 实例化Rpc客户端 45 46 print(" [x] Requesting fib(30)") 47 response = fibonacci_rpc.call(30) # 传入一个数,获得server端计算的斐波那契结果 48 print(" [.] Got %r" % response)
rabbitMQ_RPC_Server端代码:
1 #!/usr/bin/env python3.5 2 #_*_coding:utf-8_*_ 3 __author__ = 'ted.zhou' 4 import pika 5 import time 6 # 实例化连接 7 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) 8 9 # 实例化管道 10 channel = connection.channel() 11 12 # 声明队列,名称为rpc_queue 13 channel.queue_declare(queue='rpc_queue') 14 15 # 斐波那契计算函数 16 def fib(n): 17 if n == 0: 18 return 0 19 elif n == 1: 20 return 1 21 else: 22 return fib(n-1) + fib(n-2) 23 24 # 定义接收到消息后消息的处理方法 25 def on_request(ch, method, props, body): 26 n = int(body) # 接收到的字符串,由于要进行斐波那契处理,换算成数字 27 28 print(" [.] fib(%s)" % n) 29 response = fib(n) # 调用斐波那契函数计算返回结果 30 31 # 这里我们看到ch参数的作用了,它相当于管道,对!对!对!这个ch就是管道实例,props就是properties声明的属性,这些属性是是绑定到接收这个信息的管道上的 32 ch.basic_publish(exchange='', # 33 routing_key=props.reply_to, # 这里当exchange=''空时,routing_key参数就是指queue名称了,这个props.reply_to正是RPC的客户端在发送命令时一同发送过来的. 34 properties=pika.BasicProperties(correlation_id = props.correlation_id), # 同时把correlation_id也发送回去,代表执行命令的记录ID,这里我要尝试下如果返回去的值和RPC call端发送的值不一样的情况下,信息是不是不能返回 35 body=str(response)) #发送body,body就是斐波那契函数执行后返回的结果 36 ch.basic_ack(delivery_tag = method.delivery_tag) #这里delivery_tag 目前还不知道是啥作用,老师视频中说是消息持久化配置中接收端的代码,但是我觉得肯定不是 37 38 channel.basic_qos(prefetch_count=1) # 表示同一时刻保证程序只从rabbitMQ中获取一个消息进行处理,在未处理完的情况下,不会取第二个消息 39 channel.basic_consume(on_request, queue='rpc_queue') # 定义接收消息的方式 40 41 print(" [x] Awaiting RPC requests") 42 channel.start_consuming() # 开启这个管道接收消息的方法
在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。