HelloWorld

はじめに

RabbitMQ: メッセージを受信して??配信するもので、「郵便局」とみなすことができます。送信者と受信者はキューを介して対話します。キューのサイズは無制限であると考えられます。複數(shù)の送信者がキューにメッセージを送信でき、複數(shù)の受信者がキューからメッセージを受信することもできます。

コード

rabbitmq で使用されるプロトコルは amqp で、Python の推奨クライアントは pika

pip?install?pika?-i?https://pypi.douban.com/simple/

send.py

#?coding:?utf8
import?pika

#?建立一個連接
connection?=?pika.BlockingConnection(pika.ConnectionParameters(
???????????'localhost'))??#?連接本地的RabbitMQ服務器
channel?=?connection.channel()??#?獲得channel

他のマシン上のサーバーに接続したい場合は、ここにあるリンクがあります。 just fill in アドレスまたはホスト名を入力するだけです。

次にメッセージの送信を開始します。メッセージを受け入れるキューが存在することを確認してください。存在しない場合、rabbitMQ はメッセージを破棄します

channel.queue_declare(queue='hello')??#?在RabbitMQ中創(chuàng)建hello這個隊列
channel.basic_publish(exchange='',??#?使用默認的exchange來發(fā)送消息到隊列
??????????????????routing_key='hello',??#?發(fā)送到該隊列?hello?中
??????????????????body='Hello?World!')??#?消息內(nèi)容

connection.close()??#?關閉?同時flush

RabbitMQ はデフォルトで 1GB の空きディスク容量を必要とし、それ以外の場合は送信が失敗します。

このとき、ローカルキューhelloにメッセージが格納されています。rabbitmqctl list_queuesを使用すると、helloキューにメッセージが格納されていることを示す

receive.py

hello?1

が表示されます。最初にサーバーに接続した方が良いです。前に送信したときと同じです

#?coding:?utf8
import?pika
connection?=?pika.BlockingConnection(pika.ConnectionParameters(
???????????????'localhost'))
channel?=?connection.channel()

ワークキュー（タスクキュー）

ワークキューは、時間のかかるタスクを複數(shù)のワーカープロセスに分散するために使用されます。リソースを消費するタスクをすぐに実行するのではなく (これらのタスクが完了するまで待つ必要があります)、後で実行するようにこれらのタスクをスケジュールします。たとえば、タスクをメッセージとしてキューに送信し、ワーカー プロセスを開始してタスクを受け入れ、最終的に実行します。また、複數(shù)のワーカー プロセスを開始して動作させることもできます。これは、http リクエストの処理ウィンドウ內(nèi)で複雑なタスクを完了すべきではない Web アプリケーションに當てはまります。

channel.queue_declare(queue='hello')??#?此處就是聲明了?來確保該隊列?hello?存在?可以多次聲明?這里主要是為了防止接受程序先運行時出錯

def?callback(ch,?method,?properties,?body):??#?用于接收到消息后的回調(diào)
????print("?[x]?Received?%r"?%?body)

channel.basic_consume(callback,
??????????????????????queue='hello',??#?收指定隊列hello的消息
??????????????????????no_ack=True)??#在處理完消息后不發(fā)送ack給服務器
channel.start_consuming()??#?啟動消息接受?這會進入一個死循環(huán)

メッセージの配布方法はポーリングです。つまり、各ワーカー プロセスは同じ數(shù)のメッセージを取得します。

メッセージ確認

メッセージがワーカー プロセスに割り當てられているが、処理が完了する前にワーカー プロセスがクラッシュした場合、メッセージは失われる可能性があります。これは、rabbitmq がメッセージをワーカー プロセスに配布すると、メッセージが削除されるためです。

メッセージの損失を防ぐために、rabbitmq は ack を提供します。つまり、ワーカー プロセスがメッセージを受信して??処理した後、rabbitmq に ack を送信して、現(xiàn)時點でメッセージをキューから削除できることを Rabbitmq に通知します。ワーカー プロセスが停止し、rabbitmq が ack を受信しない場合、メッセージは他のワーカー プロセスに再配布されます。タイムアウトを設定する必要がなく、長時間かかるタスクでも処理できます。

ack はデフォルトで有効になっています。以前、ワーカー プロセスは ack:

channel.basic_publish(exchange='',
??????????????????routing_key='task_queue',
??????????????????body=message,
??????????????????properties=pika.BasicProperties(
?????????????????????delivery_mode?=?2,?#?使得消息持久化
??????????????????))

メッセージ永続性

を使用して no_ack=True

channel.basic_consume(callback,?queue='hello')??#?會啟用ack

コールバックを指定していましたが、場合によっては RabbitMQ が再起動され、メッセージが失われます。永続性はキューの作成時に設定できます:

(キューの性質(zhì)は一度決定すると変更できません)

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)??#?發(fā)送ack

同時に、メッセージの送信時にメッセージの永続性屬性も設定する必要があります:

channel.basic_publish (exchange='',

channel.queue_declare(queue='task_queue',?durable=True)

ただし、RabbitMQ がメッセージを受信したばかりで、それを保存する時間がなかった場合でも、メッセージは失われます。同時に、RabbitMQ は受信したすべてのメッセージを保存しません。より完全な保証が必要な場合は、発行者確認を使用する必要があります。

公平なメッセージ配信

ポーリング モードでのメッセージ配信は公平ではない可能性があります。たとえば、奇數(shù)のメッセージが重いタスクである場合、一部のプロセスは常に重いタスクを?qū)g行します。たとえば、特定のワーカー プロセスにバックログのメッセージがある場合でも、多くの ack が送信されませんが、RabbitMQ は引き続きメッセージを受信プロセスに追加できます:

??????????????????routing_key="task_queue",
??????????????????body=message,
??????????????????properties=pika.BasicProperties(
?????????????????????delivery_mode?=?2,?#?make?message?persistent
??????????????????))

を RabbitMQ に通知します。ワーカー プロセスが ack を返さない場合、それ以上のメッセージは割り當てられないことを示します。

グループ

通常、メッセージは複數(shù)のプロセスに送信されてから完了することがあります。同時に:

exchange

送信者はメッセージをキューに直接送信しますか? 実際、送信者はメッセージを交換に送信することしかできません。一方では、交換機はプロデューサーからメッセージを受信し、他方ではメッセージをキューにプッシュします。メッセージを受信したときに何をする必要があるか、メッセージを特別なメッセージに追加する必要があるかどうかを知る必要があります。 Exchange には、直接、トピック、ヘッダー、ファンアウトなどのタイプがあり、メッセージを発行するときに使用されるのはファンアウトです。これは、Exchange の値を意味します。デフォルトの Exchange を使用します

channel.basic_qos(prefetch_count=1)

一時キュー

channel.exchange_declare(exchange='logs',?type='fanout')??#?該exchange會把消息發(fā)送給所有它知道的隊列中

このように、result.method.queue はキュー名であり、

Exchange と queue

result?=?channel.queue_declare()??#?創(chuàng)建一個隨機隊列
result?=?channel.queue_declare(exclusive=True)??#?創(chuàng)建一個隨機隊列，同時在沒有接收者連接該隊列后則銷毀它
queue_name?=?result.method.queue

のログをバインドしてコピーを送信します。メッセージを送信するときは、こんにちは

メッセージを送信するときは、新しく作成されたログ交換を使用します

channel.queue_bind(exchange='logs',
???????????????queue='hello')

ルーティング

以前はバインドを使用して、交換とキューの関係を確立しました（キューは交換からのメッセージに関心があります） )、バインドするときに routing_key オプションを指定することもできます

直接交換

を使用して、ルーティング キーに対応するメッセージを同じルーティング キーにバインドされたキューに送信します

???channel.basic_publish(exchange='logs',
??????????????????routing_key='',
??????????????????body=message)

。異なる重大度のメッセージをパブリッシュする送信関數(shù):

。

channel.exchange_declare(exchange='direct_logs',
?????????????????????type='direct')

受信関數(shù)で対応する重大度をバインドします:

channel.basic_publish(exchange='direct_logs',
??????????????????routing_key=severity,
??????????????????body=message)

トピック交換を使用します

前に使用した直接交換は 1 つのルーティング キーのみをバインドできます。これを使用して、ルーティング キーのトピック交換を開くことができます。例:

"stock.usd.nyse"?"nyse.vmw"

和direct exchange一樣，在接受者那邊綁定的key與發(fā)送時指定的routing key相同即可，另外有些特殊的值：

*?代表1個單詞
#?代表0個或多個單詞

如果發(fā)送者發(fā)出的routing key都是3個部分的，如：celerity.colour.species。

Q1:
*.orange.*??對應的是中間的colour都為orange的

Q2:
*.*.rabbit??對應的是最后部分的species為rabbit的
lazy.#??????對應的是第一部分是lazy的

qucik.orange.rabbit Q1 Q2都可接收到，quick.orange.fox 只有Q1能接受到，對于lazy.pink.rabbit雖然匹配到了Q2兩次，但是只會發(fā)送一次。如果綁定時直接綁定#，則會收到所有的。

　RPC

在遠程機器上運行一個函數(shù)然后獲得結(jié)果。

１、客戶端啟動　同時設置一個臨時隊列用于接受回調(diào)，綁定該隊列

????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)

２、客戶端發(fā)送rpc請求，同時附帶reply_to對應回調(diào)隊列，correlation_id設置為每個請求的唯一id（雖然說可以為每一次RPC請求都創(chuàng)建一個回調(diào)隊列，但是這樣效率不高，如果一個客戶端只使用一個隊列，則需要使用correlation_id來匹配是哪個請求），之后阻塞在回調(diào)隊列直到收到回復

注意：如果收到了非法的correlation_id直接丟棄即可，因為有這種情況--服務器已經(jīng)發(fā)了響應但是還沒發(fā)ack就掛了，等一會服務器重啟了又會重新處理該任務，又發(fā)了一遍相應，但是這時那個請求已經(jīng)被處理掉了

channel.basic_publish(exchange='',
???????????????????????routing_key='rpc_queue',
???????????????????????properties=pika.BasicProperties(
?????????????????????????????reply_to?=?self.callback_queue,
?????????????????????????????correlation_id?=?self.corr_id,
?????????????????????????????),
???????????????????????body=str(n))??#?發(fā)出調(diào)用

while?self.response?is?None:??#?這邊就相當于阻塞了
????self.connection.process_data_events()??#?查看回調(diào)隊列
return?int(self.response)

3、請求會發(fā)送到rpc_queue隊列
4、RPC服務器從rpc_queue中取出，執(zhí)行，發(fā)送回復

channel.basic_consume(on_request,?queue='rpc_queue')??#?綁定?等待請求

#?處理之后：
ch.basic_publish(exchange='',
?????????????????routing_key=props.reply_to,
?????????????????properties=pika.BasicProperties(correlation_id?=?\
?????????????????????????????????????????????????????props.correlation_id),
?????????????????body=str(response))??#?發(fā)送回復到回調(diào)隊列
ch.basic_ack(delivery_tag?=?method.delivery_tag)??#?發(fā)送ack

5、客戶端從回調(diào)隊列中取出數(shù)據(jù)，檢查correlation_id，執(zhí)行相應操作

if?self.corr_id?==?props.correlation_id:
????????self.response?=?body

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?

以上がRabbitMQ クイック スタート Python チュートリアルの詳細內(nèi)容です。詳細については、PHP 中國語 Web サイトの他の関連記事を參照してください。