轉(zhuǎn)載自?https://blog.csdn.net/zhu_tianwei/article/details/53971296

下面主要從隊列、消息發(fā)送、消息接收方面了解消息傳遞過的一些可靠性處理。?
1、隊列?
消費(fèi)者是無法訂閱或者獲取不存在的MessageQueue中信息。消息被Exchange接受以后，如果沒有匹配的Queue，則會被丟棄。?
聲明一個隊列?
channel.queueDeclare(queue, durable, exclusive, autoDelete, arguments)?
durable：聲明隊列持久化?
exclusive：排他隊列，如果一個隊列被聲明為排他隊列，該隊列僅對首次聲明它的連接可見，并在連接斷開時自動刪除。這里需要注意三點(diǎn)：其一，排他隊列是基于連接可見的，同一連接的不同信道是可以同時訪問同一個連接創(chuàng)建的排他隊列的。其二，“首次”，如果一個連接已經(jīng)聲明了一個排他隊列，其他連接是不允許建立同名的排他隊列的，這個與普通隊列不同。其三，即使該隊列是持久化的，一旦連接關(guān)閉或者客戶端退出，該排他隊列都會被自動刪除的。這種隊列適用于只限于一個客戶端發(fā)送讀取消息的應(yīng)用場景。?
autoDelete：自動刪除，如果該隊列沒有任何訂閱的消費(fèi)者的話，該隊列會被自動刪除。這種隊列適用于臨時隊列。?
其他選項，channel.queueDeclarePassive:例如如果用戶僅僅想查詢某一個隊列是否已存在，如果不存在，不想建立該隊列，仍然可以調(diào)用queue.declare，只不過需要將參數(shù)passive設(shè)為true，傳給queue.declare，如果該隊列已存在，則會返回true；如果不存在，則會返回Error，但是不會創(chuàng)建新的隊列。?
2、發(fā)送消息?
1.發(fā)送消息設(shè)置?
channel.basicPublish(exchange, routingKey, mandatory, immediate, basicProperties, body);?
basicProperties：通過參數(shù)實(shí)現(xiàn)消息持久化，MessageProperties.PERSISTENT_TEXT_PLAIN

public BasicProperties(String contentType,//消息類型如：text/plainString contentEncoding,//編碼Map<String,Object> headers,Integer deliveryMode,//1:nonpersistent 2:persistentInteger priority,//優(yōu)先級String correlationId,String replyTo,//反饋隊列String expiration,//expiration到期時間String messageId,Date timestamp,String type,String userId,String appId,String clusterId)

mandatory：當(dāng)mandatory標(biāo)志位設(shè)置為true時，如果exchange根據(jù)自身類型和消息routeKey無法找到一個符合條件的queue，那么會調(diào)用basic.return方法將消息返還給生產(chǎn)者；當(dāng)mandatory設(shè)為false時，出現(xiàn)上述情形broker會直接將消息扔掉。?
immediate：當(dāng)immediate標(biāo)志位設(shè)置為true時，如果exchange在將消息route到queue(s)時發(fā)現(xiàn)對應(yīng)的queue上沒有消費(fèi)者，那么這條消息不會放入隊列中。當(dāng)與消息routeKey關(guān)聯(lián)的所有queue(一個或多個)都沒有消費(fèi)者時，該消息會通過basic.return方法返還給生產(chǎn)者。?
2.事務(wù)機(jī)制?
對事務(wù)的支持是AMQP協(xié)議的一個重要特性。假設(shè)當(dāng)生產(chǎn)者將一個持久化消息發(fā)送給服務(wù)器時，因為consume命令本身沒有任何Response返回，所以即使服務(wù)器崩潰，沒有持久化該消息，生產(chǎn)者也無法獲知該消息已經(jīng)丟失。如果此時使用事務(wù)，即通過txSelect()開啟一個事務(wù)，然后發(fā)送消息給服務(wù)器，然后通過txCommit()提交該事務(wù)，即可以保證，如果txCommit()提交了，則該消息一定會持久化，如果txCommit()還未提交即服務(wù)器崩潰，則該消息不會服務(wù)器就收。當(dāng)然Rabbit MQ也提供了txRollback()命令用于回滾某一個事務(wù)。

channel.txSelect() ... channel.txCommit() ... channel.txRollback()

3.Confirm機(jī)制?
事務(wù)機(jī)制會帶來大量的多余開銷，并會導(dǎo)致吞吐量下降 250% 。為了補(bǔ)救事務(wù)帶來的問題,引入了 confirmation 機(jī)制（即 Publisher Confirm）。如果設(shè)置channel為confirm狀態(tài)，則通過該channel發(fā)送的消息都會被分配一個唯一的ID，然后一旦該消息被正確的路由到匹配的隊列中后，服務(wù)器會返回給生產(chǎn)者一個Confirm，該Confirm包含該消息的ID，這樣生產(chǎn)者就會知道該消息已被正確分發(fā)。對于持久化消息，只有該消息被持久化后，才會返回Confirm。Confirm機(jī)制的最大優(yōu)點(diǎn)在于異步，生產(chǎn)者在發(fā)送消息以后，即可繼續(xù)執(zhí)行其他任務(wù)。而服務(wù)器返回Confirm后，會觸發(fā)生產(chǎn)者的回調(diào)函數(shù)，生產(chǎn)者在回調(diào)函數(shù)中處理Confirm信息。如果消息服務(wù)器發(fā)生異常，導(dǎo)致該消息丟失，會返回給生產(chǎn)者一個nack，表示消息已經(jīng)丟失，這樣生產(chǎn)者就可以通過重發(fā)消息，保證消息不丟失。Confirm機(jī)制在性能上要比事務(wù)優(yōu)越很多。但是Confirm機(jī)制，無法進(jìn)行回滾，就是一旦服務(wù)器崩潰，生產(chǎn)者無法得到Confirm信息，生產(chǎn)者其實(shí)本身也不知道該消息吃否已經(jīng)被持久化，只有繼續(xù)重發(fā)來保證消息不丟失，但是如果原先已經(jīng)持久化的消息，并不會被回滾，這樣隊列中就會存在兩條相同的消息，系統(tǒng)需要支持去重。可以mandatory配合實(shí)現(xiàn)消息的發(fā)送可靠性。

// confirm 異步機(jī)制 通過注冊listener，實(shí)現(xiàn)異步ack，提高性能channel.confirmSelect();channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {//失敗重發(fā)}@Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {//確認(rèn)ok}}); //confirm 同步機(jī)制 if(channel.waitForConfirms(timeout)){ //確認(rèn)ok}else{//失敗從發(fā)}

3、消息接收?
1.autoAck?
為了確保消息一定被消費(fèi)者處理，rabbitMQ提供了消息確認(rèn)功能，就是在消費(fèi)者處理完任務(wù)之后，就給服務(wù)器一個回饋，服務(wù)器就會將該消息刪除，如果消費(fèi)者超時不回饋，那么服務(wù)器將就將該消息重新發(fā)送給其他消費(fèi)者。默認(rèn)是開啟的，在消費(fèi)者端通過下面的方式開啟消息確認(rèn), 首先將autoAck自動確認(rèn)關(guān)閉，等我們的任務(wù)執(zhí)行完成之后，手動的去確認(rèn).

boolean autoAck = false; channel.basicConsume("hello", autoAck, consumer); ... QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); //確認(rèn)消息，已經(jīng)收到 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);

2.公平調(diào)度?
讓每個消費(fèi)者在同一時刻會分配一個任務(wù)。 通過channel.basicQos(prefetchCount);可以設(shè)置。?
3.exclusive?
和queue一樣，設(shè)置了true，只有第一個啟動的消費(fèi)者可用。?
channel.basicConsume(queue, autoAck, consumerTag, noLocal, exclusive, arguments, consumer)

總結(jié)

以上是生活随笔為你收集整理的RabbitMQ学习之消息可靠性及特性的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。