HelloWorld

??

RabbitMQ: ???? ?? ???? ?? "???"?? ??? ? ????. ???? ???? ???? ?? ?? ?????. ???? ??? ????? ??? ? ????. ?? ???? ???? ???? ?? ? ??? ?? ???? ????? ???? ?? ?? ????.

code

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服務(wù)器
channel?=?connection.channel()??#?獲得channel

??? ??? ????. ? ??? ?? ?? ??? ??? ????? ??? ??? ??? ???? ???.

?? hello ?? ???? ???? ????. Rabbitmqctl list_queues? ???? hello ?? ???? ???? ??? ????

channel.queue_declare(queue='hello')??#?在RabbitMQ中創(chuàng)建hello這個隊(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)閉?同時flush

? ? ? ????

receive .py

hello?1

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

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

?? ???(?? ???)

?? ???? ??? ????. ??? ?? ??? ??? ?? ?? ????? ???? ? ?????. ???? ?? ???? ??? ?? ???? ??(??? ??? ??? ??? ???? ?) ??? ????? ?? ??? ?????. ?? ?? ??? ???? ???? ???, ??? ????? ???? ?? ???? ????? ????, ?? ??? ????? ???? ??? ? ????. ?? http ?? ?? ? ??? ??? ??? ????? ? ?? ? ??????? ?????.

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

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()??#?啟動消息接受?這會進(jìn)入一個死循環(huán)

???? ???? ??? ?????. ?, ? ??? ????? ??? ?? ???? ????.

??? ??

???? ??? ????? ?????? ??? ???? ?? ??? ????? ???? ??, Rabbitmq? ??? ????? ???? ???? ???? ??? ? ????. , ???? ?????.

??? ??? ???? ?? Rabbitmq? ack? ?????. ?, ??? ????? ???? ???? ??? ? ack? Rabbitmq? ?? ? ???? ??? ??? ? ??? Rabbitmq? ????. ??. ??? ????? ???? Rabbitmq? ??? ?? ??? ???? ?? ??? ????? ??????. ?? ??? ?? ????? ??? ? ???? ?? ??? ??? ??? ????.

ack? ????? ????? ????. ???? ??? ?????? no_ack=True

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

ack:

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

??? ???

? ??? ??? ??????. RabbitMQ? ?? ???? ???? ?????. ?? ??? ? ???? ??? ? ????.

(?? ??? ?? ???? ??? ? ????.)

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

??? ??? ??

?? ?? ??? ??? ???? ?? ? ????. ?? ?? ?? ?? ???? ??? ??? ?? ?? ????? ?? ??? ??? ?????. ?? ?? ??? ?????? ???? ?? ???? ???? ??? RabbitMQ? ??? ???? ???? ?? ? ????. <.>

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

??? ????? ??? ?? ??? ??? ? ?? ???? ???? ??? RabbitMQ? ????. ??? ????? ?? ? ??? ?? ????? ???? ??? ?? ?? ????.

??

??? ?? ??? ???? ???? ?? ????? ???? ???? ?? ???? ???? ?? ?? ? ???? ????? ?? ? ????. ???? ???? ?????? ???? ???? ?? ????? ?? ???? ???? ?????. ??? Exchange??? ???? ??? ??? ??? ??, Exchange? ??, ??, ??, ???? ??? ???, ?? ???? ???? ???? ??? ???. ? ?? ??? ??? ?? ??? ???? ?? ?? ? ??????. ???? ?? ? ?? ?? '' ?,

channel.basic_qos(prefetch_count=1)

?? ???

? ?????.

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

?? ???? result.method.queue? ?? ?? ?? ? ??? ? ?? ??? ?????.

?? ? ??? ???

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

??? ???? ????. ????? ???? ?? ?

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

???? ?? ?

???

???? ??? ???????. ?, ?? ?? ?????. ?? ?????(?? ??? ???? ??? ????). ???? ? Routing_key ??? ???? ??? ?? ???? ???? ???? ?? ?? ?? ????. ??? ??? ?

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

??? ??, ?? ???? ??? ??:

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

?? ?? ???? ???? ???:

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

?? ?? ??

??? ??? ?? ??? ??? ??? ?? ???? ? ????. ??? ?? ???? ? ?? ??? ??? ? ????. ?:

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

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

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

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

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

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

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

　RPC

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

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

注意：如果收到了非法的correlation_id直接丟棄即可，因?yàn)橛羞@種情況--服務(wù)器已經(jīng)發(fā)了響應(yīng)但是還沒發(fā)ack就掛了，等一會服務(wù)器重啟了又會重新處理該任務(wù)，又發(fā)了一遍相應(yī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、請求會發(fā)送到rpc_queue隊(duì)列
4、RPC服務(wù)器從rpc_queue中取出，執(zhí)行，發(fā)送回復(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ā)送回復(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 ????? ?? ?????. ??? ??? PHP ??? ????? ?? ?? ??? ?????!