RabbitMq保證訊息可靠性之確認模式

前提

完成 SpringBoot 整合 RabbitMq 中的Topic萬用字元模式

一、更改Producer工程的application.yml檔案

spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: /
    username: username
    password: password
    publisher-confirm-type: correlated #開啟確認預設
server:
  port: 8080

二、更改ProducerTest.java檔案 ConfirmCallback

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;

@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class RabbitMqTest {

    @Resource
    private RabbitTemplate rabbitTemplate;
    
    @Test
    public void test() throws InterruptedException {
        String body = "確認模式的訊息";
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             *
             * @param correlationData 配置資訊
             * @param b 是否收到資訊
             * @param s 失敗的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if (b){
                    log.debug("傳送資訊到交換機成功!");
                } else {
                    log.debug("傳送資訊到交換機失敗:{}",s);
                   //第二次傳送 rabbitTemplate.convertAndSend("topic_exchange","item.body",body);
                }
            }
        });
        //為了達到確認訊息模式 
        //報出異常方法：【在消費端產生一個異常比如 100/0 就會觸發 ConfirmCallback】
        rabbitTemplate.convertAndSend("topic_exchange","item.body",body);
        Thread.sleep(2000);
    }
}
 

三、測試

首先執行 ProducerTest.java 單元測試,然後在啟動 ConsumerListener.java 訊息監聽器

1. 如果已經存在 topic_queue 請先刪除後再執行單元測試

四、小結

第一次傳送訊息,傳送訊息時出現異常問題,就會呼叫到 ConfirmCallback 方法,會再次進行傳送.保證訊息的可靠性,不會丟失.