ACK和NACK

當(dāng)我們使用RabbitMQ時用于網(wǎng)絡(luò)異常，業(yè)務(wù)處理異?；蛘邩I(yè)務(wù)錯誤導(dǎo)致消息無法立即消費(fèi)時，在這種情況下，傳輸中的信息將無法正常投遞——它們需要被重新投遞。Acknowledgements機(jī)制讓服務(wù)器和客戶端知道何時需要重新投遞。

當(dāng)設(shè)置autoACK=false 時，就可以使用手工ACK。 其實手工方式包括了手工ACK、手工NACK。

• 手工 ACK 時，會發(fā)送給Broker一個應(yīng)答，代表消息處理成功，Broker就可回送響應(yīng)給Pro；
• NACK 則表示消息處理失敗，如果設(shè)置了重回隊列，Broker端就會將沒有成功處理的消息重新發(fā)送。

使用方式

1、basicNack

Consumer消費(fèi)時，若由于業(yè)務(wù)異常，可手工 NACK 記錄日志，然后進(jìn)行補(bǔ)償: basic.nack方法為不確認(rèn)deliveryTag對應(yīng)的消息，第二個參數(shù)是否應(yīng)用于多消息，第三個參數(shù)是否requeue，如果requeue 參數(shù)設(shè)置為true ，則RabbitMQ 會重新將這條消息存入隊列，以便可以發(fā)送給下一個訂閱的消費(fèi)者, 如果requeue 參數(shù)設(shè)置為false ，則RabbitMQ立即會把消息從隊列中移除，而不會把它發(fā)送給新的消費(fèi)者。與basic.reject區(qū)別就是同時支持多個消息，可以nack該消費(fèi)者先前接收未ack的所有消息。nack后的消息也會被自己消費(fèi)到。

multiple 參數(shù)設(shè)置為false 則表示拒絕編號為deliveryTag的這一條消息，這時候basicNack 和basicReject 方法一樣;multiple 參數(shù)設(shè)置為true 則表示拒絕deliveryTag 編號之前所有未被當(dāng)前消費(fèi)者確認(rèn)的消息。

• 參數(shù)1： 消息
• 參數(shù)2： 是否應(yīng)用于多消息
• 參數(shù)3： 是否重新放回隊列，否則丟棄或者進(jìn)入死信隊列

void basicNack(long deliveryTag, boolean multiple, boolean requeue)

2、basicAck

如果由于服務(wù)器宕機(jī)等嚴(yán)重問題，就需要手工 ACK 保障Con消費(fèi)成功:deliveryTag ： 每次接收消息+1，可以做此消息處理通道的名字。

void basicAck(long deliveryTag, boolean multiple)

3、basicReject

basic.Reject拒絕deliveryTag對應(yīng)的消息，第二個參數(shù)是否requeue，true則重新入隊列，否則丟棄或者進(jìn)入死信隊列。該方法reject后，該消費(fèi)者還是會消費(fèi)到該條被reject的消息。reject一次只能拒絕一條消息，如果想要批量拒絕消息，則可以使用Basic.Nack 這個命令。

• 參數(shù)1： 消息
• 參數(shù)2： 是否重新放回隊列，否則丟棄或者進(jìn)入死信隊列

void basicReject(long deliveryTag, boolean requeue);

4、basicRecover

basic.recover是否恢復(fù)消息到隊列，參數(shù)是是否requeue，true則重新入隊列，并且盡可能的將之前recover的消息投遞給其他消費(fèi)者消費(fèi)，而不是自己再次消費(fèi)。false則消息會重新被投遞給自己。

channel.basicRecover(true);

消費(fèi)端的重回隊列

重回隊列針對沒有處理成功的消息，將消息重新投遞給Broker。
重回隊列會把消費(fèi)失敗的消息重新添加到隊列尾端，供Consumer重新消費(fèi)。
一般在實際應(yīng)用中，都會關(guān)閉重回隊列，即設(shè)置為false。

在確認(rèn)ack后再把消息通過basicPublish到隊列尾部，防止正常消息堆積。

channel.basicPublish(message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,JSON.toJSONBytes(new Object()));

實際運(yùn)用

@RabbitListener(queues = "${rabbitmq.one-queue}",
        containerFactory = "rabbitListenerContainerFactory")
@RabbitHandler
public void doRmOnduty(Message message, Channel channel) {
    try {
        String body = new String(message.getBody());
        //log.info("消息處理:{}",body);
    } catch (Exception e) {
        //消費(fèi)異常，設(shè)置重回隊列
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    } finally {
        //最終確認(rèn)消息消費(fèi)成功
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

ACK機(jī)制可以保證Con拉取到了消息，若處理失敗了，則隊列中還有這個消息，仍然可以給Con處理。ack機(jī)制是 Con 告訴 Broker 當(dāng)前消息是否成功消費(fèi)，至于 Broker 如何處理 NACK，取決于 Con 是否設(shè)置了 requeue：如果 requeue=false， 則NACK 后 Broker 還是會刪除消息的。

但一般處理消息失敗都是因為代碼邏輯出bug，即使隊列中后來仍然保留該消息，然后再給Con消費(fèi)，依舊報錯。 當(dāng)然，若一臺機(jī)器宕機(jī)，消息還有，還可以給另外機(jī)器消費(fèi)，這種情景下 ACK 很有用。

如果不使用 ACK 機(jī)制，直接把出錯消息存庫，便于日后查bug或重新執(zhí)行。

以上就是RabbitMQ消費(fèi)端ACK NACK及重回隊列機(jī)制詳解的詳細(xì)內(nèi)容，更多關(guān)于RabbitMQ消費(fèi)端ACK NACK的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論