面試題：Rabbitmq怎么保證消息的可靠性?

1.消費端消息可靠性保證：

1. 消息確認（Acknowledgements）：

消費者在接收到消息后，默認情況下RabbitMQ會自動確認消息（autoAck=true）。為保證消息可靠性，可以設(shè)置autoAck=false，使得消費者在處理完消息后手動發(fā)送確認（basicAck）。如果消費者在處理過程中發(fā)生異?；蛘呶赐瓿商幚砭徒K止運行，那么消息在超時時間內(nèi)將不會被刪除，會再次被RabbitMQ投遞給其他消費者。

2.死信隊列（Dead Letter Queue）：

當(dāng)消息不能被正常消費時（比如達到最大重試次數(shù)），可以通過設(shè)置TTL（Time To Live）或者死信交換器（Dead Letter Exchange）將消息路由至死信隊列，從而有機會后續(xù)分析和處理這些無法正常消費的消息。

2.生產(chǎn)端消息可靠性保證：

1. 消息持久化：

當(dāng)生產(chǎn)者發(fā)布消息時,可以選擇將其標(biāo)記為持久化（persistent).這意味著即使 RabbitMQ 服務(wù)器重啟，消息也不會丟失，因為它們會被存儲在磁盤上。

1. 確認（Confirm）機制：

開啟confirm回調(diào)模式后，RabbitMQ會在消息成功寫入到磁盤并至少被一個交換器接受后，向生產(chǎn)者發(fā)送一個確認（acknowledgement）。若消息丟失或無法投遞給任何隊列，RabbitMQ將會發(fā)送一個否定確認（nack). 生產(chǎn)者可以根據(jù)這些確認信號判斷消息是否成功送達并采取相應(yīng)的重試策略。

RabbitMQ作為消息中間件并啟用publisher confirms（發(fā)布者確認）與publisher returns（發(fā)布者退回）機制時，可以確保消息從生產(chǎn)者到交換機的投遞過程得到更準(zhǔn)確的狀態(tài)反饋。

1.@PostConstruct注解

@PostConstruct注解是Java EE規(guī)范中的一部分，主要用于標(biāo)記在一個Bean初始化完成后需要執(zhí)行的方法。這個注解由JSR-250定義，并且在Spring框架以及其他遵循Java EE標(biāo)準(zhǔn)的應(yīng)用服務(wù)器中廣泛支持。

**功能與用途：初始化方法,**當(dāng)容器完成對Bean的實例化并且所有依賴注入完成后，將會自動調(diào)用標(biāo)有@PostConstruct注解的方法。這為開發(fā)者提供了一個機會，在對象正式投入使用之前進行一些必要的初始化工作，比如初始化資源、預(yù)計算某些值、啟動后臺任務(wù)等增強。

2. Publisher Confirms（發(fā)布者確認）

作用： Publisher Confirm機制允許RabbitMQ服務(wù)器通知生產(chǎn)者一個消息是否已經(jīng)被交換機正確接收。當(dāng)publisher-confirm-type設(shè)置為CORRELATED時，RabbitMQ會向生產(chǎn)者發(fā)送確認或否定響應(yīng)，確認消息已到達交換機，但不保證消息已被路由到至少一個隊列中。

生產(chǎn)者到交換機的確認（消息到達交換機）

2.1.配置：

spring.rabbitmq.publisher-confirm-type = CORRELATED

2.2. 代碼實現(xiàn)

只要到達交換機就會觸發(fā)

rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            // 消息成功投遞成功并被確認
        } else {
            // 消息未能正確投遞
        }
    }
});

3.Publisher Returns（發(fā)布者退回）

作用： Publisher Return機制用于當(dāng)消息無法按照路由鍵規(guī)則路由到任何隊列時，或者由于其他原因（例如隊列滿、消息過大等）而被交換機拒絕時，RabbitMQ將消息返回給生產(chǎn)者。

交換機到隊列的確認（消息是否正常發(fā)送到了隊列）

通過實現(xiàn) ReturnCallback 接口，發(fā)送消息失敗返回，比如交換機路由不到隊列時觸發(fā)回調(diào)：

1.只有消息沒有路由到隊列的時候，才觸發(fā)該回調(diào) .

2.只要有一個隊列接受到消息了，它就認為成功.文章來源地址http://www.zghlxwxcb.cn/news/detail-850953.html

3.1 配置

spring.rabbitmq.publisher-returns = true

3.2 代碼實現(xiàn)

rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        // 處理未被路由或因某種原因被退回的消息
    }
});

4.完整代碼

@Slf4j
@Service
public class DirectProvider implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    //注解是專門數(shù)據(jù)初始化的注解, 只有其他組件注入后,初始化方法才會執(zhí)行,
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    public  void  send(OrderingOk orderingOk){

        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        System.out.println("callbackSender UUID: " + correlationData.getId());
        rabbitTemplate.convertAndSend(
                "Direct_E01", 
                "test",orderingOk,
                m-> m,correlationData);
    }

    /**
     * 確認消息是否被交換機接收。
     *
     * @param correlationData 包含消息相關(guān)數(shù)據(jù)的對象，用于識別消息的唯一性。
     * @param ack 表示消息是否被交換機確認接收。
     * @param cause 如果消息未被接收，提供未接收的原因。
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // 獲取相關(guān)數(shù)據(jù)的ID，如果相關(guān)數(shù)據(jù)為空，則設(shè)置為空字符串
        String id = correlationData != null ? correlationData.getId() : "";
        // 如果消息被確認接收，則記錄日志
        if (ack) {
            log.info("交換機已經(jīng)收到了ID為:{}的消息", id);
        } else {
            // 如果消息未被確認接收，則記錄包括未接收原因的日志
            log.info("交換機還未收到ID為:{}的消息,由于原因:{}", id, cause);
        }

    }

    /**
     * 記錄被交換機退回的消息信息。
     *
     * @param message 消息對象，包含消息體。
     * @param replyCode 返回的響應(yīng)代碼，用于指示退回的原因。
     * @param replyText 返回的響應(yīng)文本，提供關(guān)于退回的詳細信息。
     * @param exchange 退回時涉及的交換機名稱。
     * @param routingKey 退回時使用的路由鍵。
     */
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        // 記錄消息退回的日志信息
        log.info("消息{},被交換機:{}退回,原因是:{},路由key是:{}", new String(message.getBody()), exchange, replyCode, replyText, routingKey);
    }

到了這里，關(guān)于RabbitMQ消息的可靠性的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！