ACK和NACK

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

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

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

使用方式

1、basicNack

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

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

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

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

2、basicAck

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

void basicAck(long deliveryTag, boolean multiple)

3、basicReject

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

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

void basicReject(long deliveryTag, boolean requeue);

4、basicRecover

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

channel.basicRecover(true);

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

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

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

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

實(shí)際運(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è)置重回隊(duì)列
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    } finally {
        //最終確認(rèn)消息消費(fèi)成功
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

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

但一般處理消息失敗都是因?yàn)榇a邏輯出bug，即使隊(duì)列中后來(lái)仍然保留該消息，然后再給Con消費(fèi)，依舊報(bào)錯(cuò)。 當(dāng)然，若一臺(tái)機(jī)器宕機(jī)，消息還有，還可以給另外機(jī)器消費(fèi)，這種情景下 ACK 很有用。

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

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

最新評(píng)論