Python は、RabbitMQ サーバーのメッセージ キューのリモート結果を返します。

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 サイトに注目してください。