一、消息確認(rèn)機制
成都創(chuàng)新互聯(lián)公司成立于2013年,先為彰武等服務(wù)建站,彰武等地企業(yè),進(jìn)行企業(yè)商務(wù)咨詢服務(wù)。為彰武企業(yè)網(wǎng)站制作PC+手機+微官網(wǎng)三網(wǎng)同步一站式服務(wù)解決您的所有建站問題。
rabbitmq在發(fā)送消息后立即從內(nèi)存中刪除消息,因此如果消費者處理消息耗時較長,在處理過程中消費者被kill,則處理中的消息、以及其他發(fā)往該消費者的消息都將丟失。
為了保證消息不丟失,rabbitmq支持消息確認(rèn)機制,消費者可以發(fā)送ack告訴rabbitm指定消息已經(jīng)收到并處理,因此rabbitmq可以刪除該消息。
如果消費者死掉(channel關(guān)閉、connection關(guān)閉、或者TCP connection丟失),導(dǎo)致rabbitmq沒有收到ack,rabbitmq將把消息重入隊列。
不存在消息超時,這意味著處理一個消息非常長的時間也是ok的。
消息確認(rèn)機制默認(rèn)是開啟的,通過在channel.basic_consume中設(shè)置no_ack=True關(guān)閉。
注意消費者在處理消息后,不要忘記調(diào)用channel.basic_ack進(jìn)行消息確認(rèn),否則rabbitmq將不斷消耗內(nèi)存把消息重入隊列。
二、隊列/消息持久化
為了防止rabbitmq服務(wù)終止導(dǎo)致隊列和消息丟失,需要將隊列和消息標(biāo)記為持久化的:
確保rabbitmq永遠(yuǎn)不丟失隊列,需要將隊列 聲明為持久化的:
channel.queue_declare(queue='task_queue', durable=True)
將消息聲明為持久化的:
channel.basic_publish(exchange='', routing_key="task_queue", body=message, properties=pika.BasicProperties( delivery_mode = 2, # make message persistent ))
注意:盡管已經(jīng)很健壯了,但是仍然無法完全保證消息不會丟失,例如rabbitmq接收消息但是還沒有保存到硬盤的情況。
三、exchange
簡單的說,exchange的一端接收消息,另一端把消息放進(jìn)隊列。
在rabbitmq中生產(chǎn)者不會將請求直接發(fā)送給消費者,生產(chǎn)值只會把消息發(fā)給exchange,exchange收到消息后需要知道怎么做:添加到特定隊列、添加到多個隊列、還是丟棄。
exchange的類型包括direct,topic,headers,fanout
四、綁定
exchange和queue之間的聯(lián)系被稱為綁定(binding),可以簡單的看:隊列對于特定exchange上的消息感興趣
channel.queue_bind(exchange='logs', queue=result.method.queue)
此時'logs' exchange將添加消息到指定queue
綁定可以使用一個額外的routing_key參數(shù),例如:
channel.queue_bind(exchange=exchange_name, queue=queue_name, routing_key='black')
對于fanout類型的exchange來說,routing_key參數(shù)是被忽略的
五、topic exchange
發(fā)往topic exchange的消息不能攜帶任意的routing_key,必須是以點隔開的一串字符,最大255個字節(jié)
binding key也必須是相同的形式,注意binding key有兩個重要的特殊情況:
* 可以替代一個單詞
#可以替代零個或多個單詞
例如,如果binding key是*.orange.*,則可以匹配所有
如果binding key是lazy.#,則類似于帶有l(wèi)azy.orange.male.rabbit的key的消息可以匹配。
topic exchange非常強大,通過匹配routing_key可以表現(xiàn)的像存在多個exchange
六、RPC
為了接收響應(yīng),客戶端需要在發(fā)送請求時附加發(fā)送回調(diào)隊列地址:
result = channel.queue_declare(exclusive=True) callback_queue = result.method.queue channel.basic_publish(exchange='', routing_key='rpc_queue', properties=pika.BasicProperties( reply_to = callback_queue, ), body=request) # ... and some code to read a response message from the callback_queue ...