Python は RabbitMQ サーバーを操作してメッセージキュールーティングを実装します-Python チュートリアル-php.cn

Python は RabbitMQ サーバーを操作してメッセージキュールーティングを実装します

高洛峰

リリース： 2017-03-01 14:03:11

オリジナル

1289 人が閲覧しました

RabbitMQ はメッセージキューサーバーです。ここでは、Python+Pika+RabbitMQ のサーバー側環境を見て、Python を使用して RabbitMQ サーバーを操作してメッセージキューのルーティング機能を実装する方法を見ていきます

Python は Pika ライブラリを使用します。 (インストール: sudo pip install pika) RabbitMQ メッセージキューサーバーを操作できます (インストール: sudo apt-get install Rabbitmq-server) ここでは、MQ 関連のルーティング機能を見ていきます。

ルーティングキーの実装

たとえば、メッセージをすべての受信者に送信する必要があるシナリオがありますが、自由にカスタマイズする必要がある場合、一部のメッセージは一部の受信者に送信され、一部のメッセージは次の宛先に送信されます。他の受信機はどうすればいいでしょうか？この場合、ルーティングキーが使用されます。

ルーティングキーの動作原理: 各受信側のメッセージキューがスイッチにバインドされている場合、対応するルーティングキーを設定できます。送信者がスイッチを介して情報を送信する場合、ルーティングキーを指定でき、スイッチはルーティングキーに基づいてメッセージを対応するメッセージキューに送信し、受信側がメッセージを受信できるようにします。

前の記事に引き続き、send.py と accept.py を使用してルーティングキーの機能をシミュレートします。 send.py は送信側を表し、receive.py は受信側を表します。インスタンスの機能は、情報、警告、エラーの 3 つのレベルの情報をさまざまな受信側に送信することです。

send.py コード分析

#!/usr/bin/env python
#coding=utf8
import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        &#39;localhost&#39;))
channel = connection.channel()
 
#定义交换机，设置类型为direct
channel.exchange_declare(exchange=&#39;messages&#39;, type=&#39;direct&#39;)
 
#定义三个路由键
routings = [&#39;info&#39;, &#39;warning&#39;, &#39;error&#39;]
 
#将消息依次发送到交换机，并设置路由键
for routing in routings:
  message = &#39;%s message.&#39; % routing
  channel.basic_publish(exchange=&#39;messages&#39;,
             routing_key=routing,
             body=message)
  print message
 
connection.close()

ログイン後にコピー

receive.py コード分析

#!/usr/bin/env python
#coding=utf8
import pika, sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        &#39;localhost&#39;))
channel = connection.channel()
 
#定义交换机，设置类型为direct
channel.exchange_declare(exchange=&#39;messages&#39;, type=&#39;direct&#39;)
 
#从命令行获取路由键参数，如果没有，则设置为info
routings = sys.argv[1:]
if not routings:
  routings = [&#39;info&#39;]
 
#生成临时队列，并绑定到交换机上，设置路由键
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
for routing in routings:
  channel.queue_bind(exchange=&#39;messages&#39;,
            queue=queue_name,
            routing_key=routing)
 
def callback(ch, method, properties, body):
  print " [x] Received %r" % (body,)
 
channel.basic_consume(callback, queue=queue_name, no_ack=True)
 
print &#39; [*] Waiting for messages. To exit press CTRL+C&#39;
channel.start_consuming()

ログイン後にコピー

2 つのターミナルを開き、1 つはコード python accept.py info warning を実行します。これは、情報と警告メッセージのみが表示されることを意味します。受け取った。別の端末で send.py を実行すると、受信端末が情報と警告メッセージのみを受信することがわかります。複数のターミナルを開いてreceive.pyを実行し、異なるルーティングキーパラメータを渡すと、より明らかな効果が見られます。

受信側が実行されている場合、rabbitmqctl list_bindings を使用してバインディングステータスを表示できます。

ルーティングキーのファジーマッチング
ルーティングキーのファジーマッチングとは、一般的に使用される正規表現とは異なる正規表現を使用できることを意味します。ここで、「#」はすべてを意味し、「*」は 1 つの単語のみに一致します。例を読めば理解できます。

上記の例に従って、ルーティングキーのファジーマッチング機能を実装するために、引き続き send.py と accept.py を使用します。 send.py は送信側を表し、receive.py は受信側を表します。この例の機能は大まかに次のとおりです。たとえば、親しい友人がいる場合、幸せ、悲しみ、仕事、人生について話すことができます。また、楽しいことを共有できる友人もいます。幸せなことを共有できる友達がいます。不幸なことについて彼女に話すことができます。

send.py コード分析

ルーティングキーのあいまい一致が必要なため、スイッチタイプをトピックに設定する必要があります。トピックに設定されている場合は、#、* の一致記号を使用できます。

#!/usr/bin/env python
#coding=utf8
import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        &#39;localhost&#39;))
channel = connection.channel()
 
#定义交换机，设置类型为topic
channel.exchange_declare(exchange=&#39;messages&#39;, type=&#39;topic&#39;)
 
#定义路由键
routings = [&#39;happy.work&#39;, &#39;happy.life&#39;, &#39;sad.work&#39;, &#39;sad.life&#39;]
 
#将消息依次发送到交换机，并设定路由键
for routing in routings:
  message = &#39;%s message.&#39; % routing
  channel.basic_publish(exchange=&#39;messages&#39;,
             routing_key=routing,
             body=message)
  print message
 
connection.close()

ログイン後にコピー

上記の例では、4 種類のメッセージが定義されているため、説明は省略します。

receive.py コード分析

同様に、スイッチのタイプをトピックに設定する必要があります。コマンドラインからパラメータを受け取る機能が若干調整されました。つまり、パラメータがない場合はエラーで終了します。

#!/usr/bin/env python
#coding=utf8
import pika, sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        &#39;localhost&#39;))
channel = connection.channel()
 
#定义交换机，设置类型为topic
channel.exchange_declare(exchange=&#39;messages&#39;, type=&#39;topic&#39;)
 
#从命令行获取路由参数，如果没有，则报错退出
routings = sys.argv[1:]
if not routings:
  print >> sys.stderr, "Usage: %s [routing_key]..." % (sys.argv[0],)
  exit()
 
#生成临时队列，并绑定到交换机上，设置路由键
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
for routing in routings:
  channel.queue_bind(exchange=&#39;messages&#39;,
            queue=queue_name,
            routing_key=routing)
 
def callback(ch, method, properties, body):
  print " [x] Received %r" % (body,)
 
channel.basic_consume(callback, queue=queue_name, no_ack=True)
 
print &#39; [*] Waiting for messages. To exit press CTRL+C&#39;
channel.start_consuming()

ログイン後にコピー

4 つのターミナルを開き、1 つは次のように実行され、彼女と何でも話すことができることを示します。