http://www.01happy.com/python-rabbitmq-rfc-correlation-id/

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

Python 8年前 (2012-11-30) 8676浏览 3评论

上一遍演示了远程结果返回的示例，但是有一个没有提到，就是correlation id，这个是个什么东东呢？
假设有多个计算节点，控制中心开启多个线程，往这些计算节点发送数字，要求计算结果并返回，但是控制中心只开启了一个队列，所有线程都是从这个队列里获取消息，每个线程如何确定收到的消息就是该线程对应的呢？这个就是correlation id的用处了。correlation翻译成中文就是相互关联，也表达了这个意思。
correlation id运行原理：控制中心发送计算请求时设置correlation id，而后计算节点将计算结果，连同接收到的correlation id一起返回，这样控制中心就能通过correlation id来标识请求。其实correlation id也可以理解为请求的唯一标识码。
示例内容：控制中心开启多个线程，每个线程都发起一次计算请求，通过correlation id，每个线程都能准确收到相应的计算结果。
compute.py代码分析
和上面一篇相比，只需修改一个地方：将计算结果发送回控制中心时，增加参数correlation_id的设定，该参数的值其实是从控制中心发送过来的，这里只是再次发送回去。代码如下：

  1. #!/usr/bin/env python
  2. #coding=utf8
  3. import pika
  5. #连接rabbitmq服务器
  6. connection = pika.BlockingConnection(pika.ConnectionParameters(
  7.         host='localhost'))
  8. channel = connection.channel()
  10. #定义队列
  11. channel.queue_declare(queue='compute_queue')
  12. print ' [*] Waiting for n'
  14. #将n值加1
  15. def increase(n):
  16.     return n + 1
  18. #定义接收到消息的处理方法
  19. def request(ch, method, props, body):
  20.     print " [.] increase(%s)"  % (body,)
  22.     response = increase(int(body))
  24.     #将计算结果发送回控制中心，增加correlation_id的设定
  25.     ch.basic_publish(exchange='',
  26.                      routing_key=props.reply_to,
  27.                      properties=pika.BasicProperties(correlation_id = \
  28.                                                      props.correlation_id),
  29.                      body=str(response))
  30.     ch.basic_ack(delivery_tag = method.delivery_tag)
  32. channel.basic_qos(prefetch_count=1)
  33. channel.basic_consume(request, queue='compute_queue')
  35. channel.start_consuming()

center.py代码分析
控制中心代码稍微复杂些，其中比较关键的有三个地方：

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

代码如下：

  1. #!/usr/bin/env python
  2. #coding=utf8
  3. import pika, threading, uuid
  5. #自定义线程类，继承threading.Thread
  6. class MyThread(threading.Thread):
  7.     def __init__(self, func, num):
  8.         super(MyThread, self).__init__()
  9.         self.func = func
  10.         self.num = num
  12.     def run(self):
  13.         print " [x] Requesting increase(%d)" % self.num
  14.         response = self.func(self.num)
  15.         print " [.] increase(%d)=%d" % (self.num, response)
  17. #控制中心类
  18. class Center(object):
  19.     def __init__(self):
  20.         self.connection = pika.BlockingConnection(pika.ConnectionParameters(
  21.                 host='localhost'))
  23.         self.channel = self.connection.channel()
  25.         #定义接收返回消息的队列
  26.         result = self.channel.queue_declare(exclusive=True)
  27.         self.callback_queue = result.method.queue
  29.         self.channel.basic_consume(self.on_response,
  30.                                    no_ack=True,
  31.                                    queue=self.callback_queue)
  33.         #返回的结果都会存储在该字典里
  34.         self.response = {}
  36.     #定义接收到返回消息的处理方法
  37.     def on_response(self, ch, method, props, body):
  38.         self.response[props.correlation_id] = body
  40.     def request(self, n):
  41.         corr_id = str(uuid.uuid4())
  42.         self.response[corr_id] = None
  44.         #发送计算请求，并设定返回队列和correlation_id
  45.         self.channel.basic_publish(exchange='',
  46.                                    routing_key='compute_queue',
  47.                                    properties=pika.BasicProperties(
  48.                                          reply_to = self.callback_queue,
  49.                                          correlation_id = corr_id,
  50.                                          ),
  51.                                    body=str(n))
  52.         #接收返回的数据
  53.         while self.response[corr_id] is None:
  54.             self.connection.process_data_events()
  55.         return int(self.response[corr_id])
  57. center = Center()
  58. #发起5次计算请求
  59. nums= [10, 20, 30, 40 ,50]
  60. threads = []
  61. for num in nums:
  62.     threads.append(MyThread(center.request, num))
  63. for thread in threads:
  64.     thread.start()
  65. for thread in threads:
  66.     thread.join()

笔者开启了两个终端，来运行compute.py，开启一个终端来运行center.py，最后结果输出截图如下：
python - rabbitmq - pika - 图1
python使用rabbitmq多节点结果返回图示
可以看到虽然获取的结果不是顺序输出，但是结果和源数据都是对应的。
这边示例的做法就是创建一个队列，使用correlation id来标识每次请求。也有做法可以不使用correlation id，就是每请求一次，就创建一个临时队列，不过这样太消耗性能了，官方也不推荐这么做。
python - rabbitmq - pika - 图2

转载请注明：快乐编程 » python使用rabbitmq实例七，相互关联编号correlation id
继续浏览有关 rabbitmq 的文章

分享到

1.
上一篇 python使用rabbitmq实例六，远程结果返回 python采用pika库使用rabbitmq总结，多篇笔记和示例 下一篇

与本文相关的文章