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

上一遍演示了远程结果返回的示例，但是有一个没有提到，就是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代码分析

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

使用python的uuid来产生唯一的correlation_id。

发送计算请求时，设定参数correlation_id。

定义一个字典来保存返回的数据，并且键值为相应线程产生的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多节点结果返回图示

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

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

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

RabbitMQ（六）：回调队列callback queue、关联标识correlation id、实现简单的RPC系统
博客翻译自:RabbitMQ Tutorials Java版 RabbitMQ(一):Hello World程序 RabbitMQ(二):Work Queues.循环分发.消息确认.持久化.公平分发 ...
python rabitmq_python使用rabbitmq实例二，工作队列
上一篇介绍了rabbitmq的安装和经典的hello world!实例.这里将对工作队列(Work Queues)做一个了解.因为是接上一篇说明的,所以如果没看过上一篇,看这篇可能会比较难理解.上一篇 ...
RabbitMQ（七）：常用方法说明与学习小结
RabbitMQ(一):Hello World程序 RabbitMQ(二):Work Queues.循环分发.消息确认.持久化.公平分发 RabbitMQ(三):Exchange交换器--fanout ...
Rabbitmq超级详细的笔记，包括安装，基本命令，rabbitmq的七种消息模式，以及死信队列，延迟队列，优先级队列和惰性队列的介绍
RabbitMQ 文章目录 RabbitMQ 1 RabbitMQ介绍 1.1 基本介绍 1.2 RabbitMQ的安装 1.2.1 ubuntu20.04 安装rabbitmq 1.2.2 cent ...
关于学习Python的一点学习总结（35-＞关联超类）
76.调用为关联的超类(继承的类)构造函数先举个例子: class Bird: def __init__(self): self.hungry = True def eat(self): if se ...
属性与内存管理(属性与内存管理都是相互关联的)
<span style="font-size:18px;"> 属性与内存管理(属性与内存管理都是相互关联的)第一部分一,属性:属性是OC2.0之后出来的新语法,用来取代 ...
python装饰器实例-python装饰器使用实例详解
这篇文章主要介绍了python装饰器使用实例详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 python装饰器的作用就是在不想改变原函数代码的情 ...
Python类属性和实例属性分别是什么？
实例属性即同一个类的不同实例,其值是互不关联的,也不会互相影响的,定义时使用"self.属性名",调用时也使用"self.属性名". 类属性则是同一个类的所有实 ...
Python 同一个类中不同函数相互调用
本文采取:Python的类中函数方法相互调用的两种方式_狗哥的博客-CSDN博客 __metaclass__=type class Stu:name=Noneage=Noneschool=" ...

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

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

最新文章

热门文章