RabbitMq消息可靠性之确认模式 通俗易懂 超详细 内含案例

Posted 2020-12-18 beixuan

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 方法,会再次进行发送.保证消息的可靠性,不会丢失.