ホームページ >バックエンド開発 >Python チュートリアル >Python は、RabbitMQ サーバーのメッセージ キューのリモート結果を返します。
Python は、RabbitMQ サーバーのメッセージ キューのリモート結果を返します。
- 高洛峰オリジナル
- 2017-03-01 14:02:031435ブラウズ
RabbitMQ は MQ に基づいたサーバーです。Python は Pika ライブラリを介してプログラム制御に使用できます。ここでは、Python による RabbitMQ サーバー メッセージ キューの操作のリモート結果の返しについて詳しく説明します。テスト環境はこちら: Ubuntu14.04 + Python 2.7.4
RabbitMQサーバーsudo apt-get install rabbitmq-server
PythonではRabbitMQを使用するにはPikaライブラリが必要です
sudo pip install pikaリモート結果リターン
メッセージ送信者はメッセージを送信し、結果は返されません。もちろんメッセージを送信するだけであれば問題ありませんが、実際には受信側で受信したメッセージを処理して送信側に返す必要が生じることが多いです。
処理方法の説明: 情報を送信する前に、送信側はメッセージを受信するための一時キューを生成します。このキューは、返された結果を受信するために使用されます。実際、ここでは受信側と送信側の概念が比較的曖昧です。送信側もメッセージを受信する必要があり、受信側もメッセージを送信する必要があるためです。そのため、ここでは別の例を使用してこのプロセスを示します。
内容の例: コントロール センターとコンピューティング ノードがあるとします。コントロール センターは、自然数 N をコンピューティング ノードに送信し、コンピューティング ノードは、N の値に 1 を加算して、それをコントロール センターに返します。ここでは、center.py はコントロール センターをシミュレートするために使用され、compute.py はコンピューティング ノードをシミュレートするために使用されます。
compute.py コード分析
#!/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, properties, body):
print " [.] increase(%s)" % (body,)
response = increase(int(body))
#将计算结果发送回控制中心
ch.basic_publish(exchange='',
routing_key=properties.reply_to,
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 コード分析
#!/usr/bin/env python
#coding=utf8
import pika
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)
#定义接收到返回消息的处理方法
def on_response(self, ch, method, props, body):
self.response = body
def request(self, n):
self.response = None
#发送计算请求，并声明返回队列
self.channel.basic_publish(exchange='',
routing_key='compute_queue',
properties=pika.BasicProperties(
reply_to = self.callback_queue,
),
body=str(n))
#接收返回的数据
while self.response is None:
self.connection.process_data_events()
return int(self.response)
center = Center()
print " [x] Requesting increase(30)"
response = center.request(30)
print " [.] Got %r" % (response,)
上記のコード例では、返されたデータを受信するためのキューと処理メソッドを定義し、リクエストを送信するときにキューを Reply_to に割り当てます。このパラメーターは計算ノードのコードで使用されます。戻りキューを取得します。
2 つのターミナルを開き、1 つはコード python compute.py を実行し、もう 1 つのターミナルは center.py を実行します。実行が成功すると、効果が確認できるはずです。
テスト中に、作者はいくつかの小さな問題に遭遇しました。つまり、center.py がメッセージを送信するときに戻りキューが指定されていなかったため、結果を計算した後にデータを送り返すときに compute.py がエラーを報告しました。 routing_key が存在しないことを示し、再度実行するとエラーが報告されました。 Rabbitmqctl list_queues を使用してキューを表示すると、compute_queue キューに 1 つのデータがあることがわかります。このデータは、compute.py が再実行されるたびに再処理されます。その後、/etc/init.d/rabbitmq-server restart を使用して Rabbitmq を再起動しましたが、すべて問題ありませんでした。
相関 ID 前回、リモート結果を返す例を示しましたが、言及されていなかったことが 1 つあります。それは相関 ID です。
複数のコンピューティング ノードがあり、コントロール センターが複数のスレッドを開始し、これらのコンピューティング ノードに数値を送信し、計算結果を要求して返したとします。ただし、コントロール センターは 1 つのキューのみを開き、すべてのスレッドがこのキューからメッセージを取得する方法を説明します。スレッドは、受信したメッセージがそのスレッドに対応するメッセージであると判断しますか?これは相関 ID の使用です。相関関係は中国語では相互相関と訳されますが、これもこの意味を表しています。
相関IDの動作原理：コントロールセンターは計算リクエストを送信するときに相関IDを設定し、計算ノードは受信した相関IDとともに計算結果を返すため、コントロールセンターは相関IDを通じてリクエストを識別できます。実際、相関 ID は、リクエストの一意の識別コードとしても理解できます。
コンテンツの例: コントロール センターは複数のスレッドを開始し、各スレッドは相関 ID を通じて計算リクエストを開始し、対応する計算結果を正確に受け取ることができます。
compute.py コード分析
前の記事と比較すると、変更が必要なのは 1 つだけです。計算結果をコントロール センターに送信するときに、パラメーター correlation_id の設定を追加します。このパラメーターの値は、実際には 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 コード分析
コントロール センターのコードは少し複雑で、重要なポイントが 3 つあります:
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 を実行するために 2 つの端末を開いて、center.py を実行するために 1 つの端末を開きます。 最終的な結果出力のスクリーンショットは次のとおりです。取得結果は順番に出力されていませんが、結果はソースデータに対応していることがわかります。
ここでの例は、キューを作成し、相関 ID を使用して各リクエストを識別することです。相関 ID を使用しない方法もあります。これは、リクエストが行われるたびに一時キューを作成します。ただし、これはパフォーマンスの消費が多すぎるため、公式には推奨されていません。
RabbitMQ サーバー メッセージ キューのリモート結果の戻りを操作する Python に関連するその他の記事については、PHP 中国語 Web サイトに注目してください。