HelloWorld

はじめに

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

コード

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

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

send.py

#?coding:?utf8
import?pika

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

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

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

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

connection.close()??#?關(guān)閉?同時(shí)flush

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

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

receive.py

hello?1

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

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

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

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

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

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

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

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

メッセージ確認(rèn)

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

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

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')??#?會(huì)啟用ack

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

(キューの性質(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

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

channel.basic_publish (exchange='',

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

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

公平なメッセージ配信

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

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

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

グループ

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

exchange

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

channel.basic_qos(prefetch_count=1)

一時(shí)キュー

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

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

Exchange と queue

result?=?channel.queue_declare()??#?創(chuàng)建一個(gè)隨機(jī)隊(duì)列
result?=?channel.queue_declare(exclusive=True)??#?創(chuàng)建一個(gè)隨機(jī)隊(duì)列，同時(shí)在沒有接收者連接該隊(duì)列后則銷毀它
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ā)送時(shí)指定的routing key相同即可，另外有些特殊的值：

*?代表1個(gè)單詞
#?代表0個(gè)或多個(gè)單詞

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

Q1:
*.orange.*??對(duì)應(yīng)的是中間的colour都為orange的

Q2:
*.*.rabbit??對(duì)應(yīng)的是最后部分的species為rabbit的
lazy.#??????對(duì)應(yīng)的是第一部分是lazy的

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

　RPC

在遠(yuǎn)程機(jī)器上運(yùn)行一個(gè)函數(shù)然后獲得結(jié)果。

１、客戶端啟動(dòng)　同時(shí)設(shè)置一個(gè)臨時(shí)隊(duì)列用于接受回調(diào)，綁定該隊(duì)列

????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請(qǐng)求，同時(shí)附帶reply_to對(duì)應(yīng)回調(diào)隊(duì)列，correlation_id設(shè)置為每個(gè)請(qǐng)求的唯一id（雖然說可以為每一次RPC請(qǐng)求都創(chuàng)建一個(gè)回調(diào)隊(duì)列，但是這樣效率不高，如果一個(gè)客戶端只使用一個(gè)隊(duì)列，則需要使用correlation_id來匹配是哪個(gè)請(qǐng)求），之后阻塞在回調(diào)隊(duì)列直到收到回復(fù)

注意：如果收到了非法的correlation_id直接丟棄即可，因?yàn)橛羞@種情況--服務(wù)器已經(jīng)發(fā)了響應(yīng)但是還沒發(fā)ack就掛了，等一會(huì)服務(wù)器重啟了又會(huì)重新處理該任務(wù)，又發(fā)了一遍相應(yīng)，但是這時(shí)那個(gè)請(qǐng)求已經(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:??#?這邊就相當(dāng)于阻塞了
????self.connection.process_data_events()??#?查看回調(diào)隊(duì)列
return?int(self.response)

3、請(qǐng)求會(huì)發(fā)送到rpc_queue隊(duì)列
4、RPC服務(wù)器從rpc_queue中取出，執(zhí)行，發(fā)送回復(fù)

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

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

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

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

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

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