基本の研究に関する
最初のレッスンに続いて、RabbitMQは
公式サイトから2番目のレッスンの翻訳を公開し
ます 。 前述のように、すべての例はpythonで記述されていますが、ほとんどの
一般的な言語で実装でき
ます 。
タスクキュー
最初のレッスンでは、2つのプログラムを作成しました。1つはメッセージを送信し、2つ目はメッセージを受信しました。 このレッスンでは、リソースを集中的に使用するタスクを複数のサブスクライバーに分散するために使用されるキューを作成します。
このようなキューの主な目的は、タスクをすぐに開始せず、完了するまで待たないことです。 代わりに、タスクが遅延します。 各メッセージは1つのタスクに対応しています。 バックグラウンドで実行されているハンドラープログラムは、処理のためにタスクを受け入れ、しばらくすると完了します。 複数のハンドラーを実行すると、タスクはそれらの間で分割されます。
この動作原理は、HTTP要求時にリソースを大量に消費するタスクを処理できないWebアプリケーションで使用する場合に特に役立ちます。
準備する
前のレッスンでは、「Hello World！」というテキストを含むメッセージを送信しました。 そして、リソースを集中的に使用するタスクに対応するメッセージを送信します。 画像のサイズ変更やPDFファイルのレンダリングなどの実際のタスクは実行しません
。time.sleep（）関数を使用してスタブを作成しましょう。 タスクの複雑さは、メッセージ行のポイントの数によって決まります。 各ポイントは1秒間「実行」されます。 たとえば、メッセージ「Hello ...」を含むタスクは3秒間実行されます。
コマンドラインから任意のメッセージを送信できるように、
前の例 のsend.pyプログラム
コードをわずかに
変更します。 このプログラムは、新しいタスクを計画して、順番にメッセージを送信します。
new_task.pyと
呼びます：
import sys message = ' '.join(sys.argv[1:]) or "Hello World!" channel.basic_publish(exchange='', routing_key='hello', body=message) print " [x] Sent %r" % (message,)
前の例の
receive.pyプログラムも変更する必要があります。メッセージテキストの各ポイントに1秒間、有用な作業の実装をシミュレートする必要があります。 プログラムはキューからメッセージを受信し、タスクを完了します。
worker.pyと
呼びます：
import time def callback(ch, method, properties, body): print " [x] Received %r" % (body,) time.sleep( body.count('.') ) print " [x] Done"
周期的分布
タスクキューを使用する利点の1つは、複数のプログラムと並行して作業できることです。 すべての着信タスクを完了する時間がない場合は、単純にハンドラーの数を追加できます。
まず、2つの
worker.pyプログラムを
一度に実行しましょう。 どちらもキューからメッセージを受信しますが、どのくらい正確ですか？ 今から見ます。
3つのターミナルウィンドウを開く必要があります。 そのうち2
つでは 、
worker.pyプログラムが起動します。 これらは、C1とC2の2つのサブスクライバーになります。
shell1$ python worker.py [*] Waiting for messages. To exit press CTRL+C
shell2$ python worker.py [*] Waiting for messages. To exit press CTRL+C
3番目のウィンドウで、新しいタスクを公開します。 サブスクライバーが開始された後、任意の数のメッセージを送信できます。
shell3$ python new_task.py First message. shell3$ python new_task.py Second message.. shell3$ python new_task.py Third message... shell3$ python new_task.py Fourth message.... shell3$ python new_task.py Fifth message.....
サブスクライバーに配信されたものを見てみましょう。
shell1$ python worker.py [*] Waiting for messages. To exit press CTRL+C [x] Received 'First message.' [x] Received 'Third message...' [x] Received 'Fifth message.....'
shell2$ python worker.py [*] Waiting for messages. To exit press CTRL+C [x] Received 'Second message..' [x] Received 'Fourth message....'
デフォルトでは、RabbitMQは新しい各メッセージを次のサブスクライバーに転送します。 したがって、すべてのサブスクライバーは同じ数のメッセージを受け取ります。 このメッセージ配信方法は、サイクリック
[ ラウンドロビンアルゴリズム]と呼ばれます。 3人以上のサブスクライバーで同じことを試してください。
メッセージ確認
これらのタスクには数秒かかります。 おそらく、ハンドラーがタスクを開始したが、予期せず動作を停止し、部分的にしか完了しなかった場合にどうなるかを既に疑問に思っているかもしれません。 プログラムの現在の実装では、メッセージはRabbitMQがサブスクライバーに配信するとすぐに削除されます。 したがって、操作中にハンドラーを停止すると、タスクは完了せず、メッセージは失われます。 まだ処理されていない配信メッセージも失われます。
しかし、タスクを失いたくありません。 あるハンドラーの緊急出口が発生した場合、メッセージは別のハンドラーに送信される必要があります。
失われたメッセージがないことを確認できるように、RabbitMQはメッセージ確認をサポートしています。 確認（
ack ）は、受信したメッセージが処理され、RabbitMQがそれを削除できることをRabbitMQに通知するために、サブスクライバーによって送信されます。
サブスクライバーが動作を停止し、確認を送信しなかった場合、RabbitMQはメッセージが処理されていないことを理解し、別のサブスクライバーに転送します。 そのため、ハンドラープログラムの実行が予期せず停止した場合でも、1つのメッセージが失われないことを確認できます。
メッセージを処理するためのタイムアウトはありません。 RabbitMQは、最初のサブスクライバーへの接続が閉じられている場合にのみ、それらを別のサブスクライバーに転送するため、メッセージの処理時間に制限はありません。
デフォルトでは、メッセージの手動確認が使用されます。 前の例では、
no_ack = Trueを指定してメッセージの自動確認を強制しました。 ここで、このフラグを削除し、タスクが完了した直後にハンドラーから確認を送信します。
def callback(ch, method, properties, body): print " [x] Received %r" % (body,) time.sleep( body.count('.') ) print " [x] Done" ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_consume(callback, queue='hello')
これで、メッセージの処理中に
Ctrl + Cを押してハンドラーを停止しても、何も失われません。 ハンドラーが停止した後、RabbitMQは未確認のメッセージを再送信します。
メッセージを確認することを忘れないでください
開発者は、コードに
basic_ackを追加するのを
忘れることがあります。 この小さなエラーの結果は重大な場合があります。 メッセージはハンドラープログラムが停止したときにのみ再送信されますが、RabbitMQはより多くのメモリを消費します。 未確認のメッセージは削除されません。
これらの種類のエラーをデバッグするには、
rabbitmqctlを使用して、
messages_unacknowledgedフィールド（未確認メッセージ）を表示できます。
$ sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged Listing queues ... hello 0 0 ...done.
[または、 最初の部分で引用したより便利な監視スクリプトを使用する]メッセージの安定性
サブスクライバーが予期せず動作を停止した場合に、タスクを失わない方法を見つけました。 ただし、RabbitMQサーバーが動作を停止すると、タスクは失われます。
デフォルトでは、RabbitMQサーバーが停止またはクラッシュすると、すべてのキューとメッセージが失われますが、この動作は変更できます。 サーバーの再起動後にメッセージがキューに残るためには、キューとメッセージの両方を安定させる必要があります。
まず、キューが失われないことを確認します。 これを行うには、それを持続可能（
永続的 ）として宣言する必要があります。
channel.queue_declare(queue='hello', durable=True)
このコマンド自体は正しいですが、
helloキューがすでに不安定であると宣言されているため、現在は機能しません。 RabbitMQでは、既存のキューのパラメーターをオーバーライドすることはできません。これを行うと、エラーが返されます。 しかし、簡単な回避策があります-別の名前、たとえば
task_queueでキューを宣言しましょう：
channel.queue_declare(queue='task_queue', durable=True)
このコードは、プロバイダープログラムとサブスクライバープログラムの両方で修正する必要があります。
したがって、RabbitMQサーバーの再起動時に
task_queueキュー
が失われないことを確認できます。 次に、メッセージを回復力があるものとしてマークする必要があります。 これを行うには、
delivery_modeプロパティに値
2を 渡します。
channel.basic_publish(exchange='', routing_key="task_queue", body=message, properties=pika.BasicProperties( delivery_mode = 2,
メッセージの安定性に関する注意
メッセージを持続可能としてマークしても、メッセージが失われないという保証はありません。 これにより、RabbitMQがメッセージをディスクに保存するように強制されるという事実にもかかわらず、RabbitMQがメッセージの受け入れを確認したがまだ記録していない短時間があります。 また、RabbitMQは各メッセージに対して
fsync（2）を実行しないため、一部のメッセージはキャッシュできますが、まだディスクに書き込まれません。 メッセージの安定性の保証は完全ではありませんが、タスクキューには十分です。 より高い信頼性が必要な場合は、トランザクションをトランザクションにラップできます。
メッセージの均一な配布
必要に応じてメッセージ配信がまだ機能しないことに気づいたかもしれません。 たとえば、2人のサブスクライバーが働いているときに、すべての奇数のメッセージに複雑なタスクが含まれている場合（
完了までにかなりの時間が必要）、単純なメッセージにも単純なタスクが含まれている場合、最初のハンドラーは常にビジー状態になり、2番目のほとんどの時間は無料になります。 しかし、RabbitMQはそれについて何も知らず、引き続きサブスクライバーにメッセージを送信します。
これは、RabbitMQがキューに入った瞬間にメッセージを配信し、サブスクライバーからの未確認メッセージの数を考慮しないためです。 RabbitMQは、n番目のサブスクライバーにn番目ごとのメッセージを送信するだけです。
この動作を変更するには、
basic_qosメソッドと
prefetch_count = 1オプションを使用します。 これにより、RabbitMQは一度に複数のメッセージをサブスクライバーに送信しなくなります。 つまり、サブスクライバーは、前のメッセージを処理して確認するまで新しいメッセージを受信しません。 RabbitMQは、最初の無料のサブスクライバーにメッセージを送信します。
channel.basic_qos(prefetch_count=1)
キューサイズに関する注意
すべてのサブスクライバがビジーの場合、キューサイズが増加する可能性があります。 これに注意を払い、おそらくサブスクライバの数を増やす必要があります。
さて、今ではすべて一緒に
new_task.pyの完全なコードは
次のとおりです 。
worker.pyの完全なコード：
メッセージ確認応答と
prefetch_countを使用して、タスクキューを作成できます。 復元力を設定すると、RabbitMQサーバーを再起動した後でもタスクを持続させることができます。
3番目のレッスンでは、1つのメッセージを複数のサブスクライバーに送信する方法を見ていきます。