最新消息:yaf表单扩展中新增加了浮点数、日期和集合的校验。php yaf框架扩展实践三——表单

python使用rabbitmq实例七,相互关联编号correlation id

Python 6526浏览 1评论

上一遍演示了远程结果返回的示例,但是有一个没有提到,就是correlation id,这个是个什么东东呢?

假设有多个计算节点,控制中心开启多个线程,往这些计算节点发送数字,要求计算结果并返回,但是控制中心只开启了一个队列,所有线程都是从这个队列里获取消息,每个线程如何确定收到的消息就是该线程对应的呢?这个就是correlation id的用处了。correlation翻译成中文就是相互关联,也表达了这个意思。

correlation id运行原理:控制中心发送计算请求时设置correlation id,而后计算节点将计算结果,连同接收到的correlation id一起返回,这样控制中心就能通过correlation id来标识请求。其实correlation id也可以理解为请求的唯一标识码。

示例内容:控制中心开启多个线程,每个线程都发起一次计算请求,通过correlation id,每个线程都能准确收到相应的计算结果。

compute.py代码分析

和上面一篇相比,只需修改一个地方:将计算结果发送回控制中心时,增加参数correlation_id的设定,该参数的值其实是从控制中心发送过来的,这里只是再次发送回去。代码如下:

#!/usr/bin/env python
#coding=utf8
import pika

#连接rabbitmq服务器
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

#定义队列
channel.queue_declare(queue='compute_queue')
print ' [*] Waiting for n'

#将n值加1
def increase(n):
    return n + 1

#定义接收到消息的处理方法
def request(ch, method, props, body):
    print " [.] increase(%s)"  % (body,)

    response = increase(int(body))

    #将计算结果发送回控制中心,增加correlation_id的设定
    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(request, queue='compute_queue')

channel.start_consuming()

center.py代码分析

控制中心代码稍微复杂些,其中比较关键的有三个地方:

  1. 使用python的uuid来产生唯一的correlation_id。
  2. 发送计算请求时,设定参数correlation_id。
  3. 定义一个字典来保存返回的数据,并且键值为相应线程产生的correlation_id。

代码如下:

#!/usr/bin/env python
#coding=utf8
import pika, threading, uuid

#自定义线程类,继承threading.Thread
class MyThread(threading.Thread):
    def __init__(self, func, num):
        super(MyThread, self).__init__()
        self.func = func
        self.num = num

    def run(self):
        print " [x] Requesting increase(%d)" % self.num
        response = self.func(self.num)
        print " [.] increase(%d)=%d" % (self.num, response)

#控制中心类
class Center(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,
                                   no_ack=True,
                                   queue=self.callback_queue)

        #返回的结果都会存储在该字典里
        self.response = {}

    #定义接收到返回消息的处理方法
    def on_response(self, ch, method, props, body):
        self.response[props.correlation_id] = body

    def request(self, n):
        corr_id = str(uuid.uuid4())
        self.response[corr_id] = None

        #发送计算请求,并设定返回队列和correlation_id
        self.channel.basic_publish(exchange='',
                                   routing_key='compute_queue',
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = corr_id,
                                         ),
                                   body=str(n))
        #接收返回的数据
        while self.response[corr_id] is None:
            self.connection.process_data_events()
        return int(self.response[corr_id])

center = Center()
#发起5次计算请求
nums= [10, 20, 30, 40 ,50]
threads = []
for num in nums:
    threads.append(MyThread(center.request, num))
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

笔者开启了两个终端,来运行compute.py,开启一个终端来运行center.py,最后结果输出截图如下:

python使用rabbitmq多节点结果返回图示

python使用rabbitmq多节点结果返回图示

可以看到虽然获取的结果不是顺序输出,但是结果和源数据都是对应的。

这边示例的做法就是创建一个队列,使用correlation id来标识每次请求。也有做法可以不使用correlation id,就是每请求一次,就创建一个临时队列,不过这样太消耗性能了,官方也不推荐这么做。

转载请注明:快乐编程 » python使用rabbitmq实例七,相互关联编号correlation id

发表我的评论
取消评论

表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

网友最新评论 (1)

  1. 请问在“#定义接收到消息的处理方法”时 def on_response(self, ch, method, props, body):
    self.response[props.correlation_id] = body

    其中的ch是什么意思?python不是很熟悉,求解释,感谢
    sebastian3年前 (2014-07-07)回复