【RabbitMQ Confirm和returns保证生产者消息可靠性】

  • A+
所属分类:java Spring Boot

RabbitMQ生产者消息可靠性

网上太多文章只说 confirm可以保证消息可靠性,经过代码实际测试压根就不靠谱。本文讲述一个更加严谨可靠的消息生产者。如果还有问题希望大家能够在评论区指出。

实验环境

软件环境和版本

  • RabbitMQ 3.9.17
  • Spring boot 2.6.5

spring boot yml 配置

spring: 
  rabbitmq:
    host: 192.168.245.131
    username: admin
    password: admin
    port: 5672
    virtual-host: /
    listener:
      simple:
        acknowledge-mode: manual # 手动ACK
    publisher-confirm-type: correlated # 开启Confirm回调
    publisher-returns: true # 开启returns回调

RabbitMQ控制台

队列为空。
【RabbitMQ Confirm和returns保证生产者消息可靠性】

创建交换机和队列并绑定

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    public static final String EXCHANGE = "agreementExchange";
    public static final String QUEUE = "agreementQueue";
    public static final String ROUTING_KEY = "*.rabbit.*";

    @Bean
    public Exchange agreementExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE).build();
    }

    @Bean
    public Queue agreementQueue(){
        return QueueBuilder.durable(QUEUE).build();
    }

    @Bean
    public Binding agreementBinding(Exchange agreementExchange,Queue agreementQueue){
        return BindingBuilder.bind(agreementQueue).to(agreementExchange).with(ROUTING_KEY).noargs();
    }
}

最基本的生产者代码

import com.stages.payment.start.config.mq.RabbitMQConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;

@SpringBootTest
class StagesPaymentStartApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() throws IOException {
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, "black.rabbit.male", "这里是消息", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                return message;
            }
        });
        System.out.println(Thread.currentThread().getName()+"程序没有中断");
    }

}

消息生产者流程梳理

消息发送流程图

  1. 发送message到Exchange交换机
  2. 交换机通过路由发送message给Queue
  3. Queue根据消息属性决定存放在磁盘还是内存

【RabbitMQ Confirm和returns保证生产者消息可靠性】

简单理解两种机制

上图我们将消息发送简单分为了三个步骤。confirm和returns可以简单理解为两个切面方法在不同的阶段进行回调。

confirm:

在第一步调用成功后触发,无论消息是否成功发送给交换机都会触发。

returns:

在第二步和第三步任何一步失败的情况下才会触发,正常情况不触发returns回调。


深入confirm机制

confirm不能保证消息的完全可靠,他只能保证消息到达交换机。只要到达交换机无论是否持久化成功或者到达队列都会返回成功。

confirm返回成功但消息丢失的情况模拟

confirm 只能保证消息成功到达交换机,那么当交换机正常,队列不存在的时候消息会丢失,而confirm返回了成功。

生产者代码:

import com.stages.payment.start.config.mq.RabbitMQConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;

@SpringBootTest
class StagesPaymentStartApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() throws IOException {
        // 成功或失败都会触发
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println(Thread.currentThread().getName()+"消息发送交换机结果: "+ack);
            }
        });
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, "black.rabbit1.male", "这里是消息4", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                messageProperties.setCorrelationId("123");
                messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                return message;
            }
        });
        System.out.println(Thread.currentThread().getName()+"程序没有中断");
    }

}

运行结果:数据丢失且程序无报错。
【RabbitMQ Confirm和returns保证生产者消息可靠性】
【RabbitMQ Confirm和returns保证生产者消息可靠性】

confirm返回失败的情况模拟

消息无法发送到交换机时confirm返回失败。当交换机不存在时返回失败。

运行结果:confirm返回失败且数据丢失
【RabbitMQ Confirm和returns保证生产者消息可靠性】
【RabbitMQ Confirm和returns保证生产者消息可靠性】

结论

单凭confirm机制根本不能保证消息生产者的可靠性。

深入returns机制

returns不能保证消息的完全可靠,他能保证消息到达队列。只要到达队列无论是否持久化都会返回成功。并且,如果消息没有到达交换机那么returns将无法触发。

持久化失败returns触发情况模拟

生产者代码:

import com.stages.payment.start.config.mq.RabbitMQConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;

@SpringBootTest
class StagesPaymentStartApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() throws IOException {
        // 成功或失败都会触发
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println(Thread.currentThread().getName()+"消息发送交换机结果: "+ack);
            }
        });
        // 发送到队列失败时才会触发
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                System.out.println(Thread.currentThread().getName()+"消息发送队列结果: "+returned);
            }
        });
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, "black.rabbit1.male", "这里是消息4", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                messageProperties.setCorrelationId("123");
                messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                return message;
            }
        });
        System.out.println(Thread.currentThread().getName()+"程序没有中断");
    }

}

运行结果:数据丢失,returns触发
【RabbitMQ Confirm和returns保证生产者消息可靠性】
【RabbitMQ Confirm和returns保证生产者消息可靠性】

confirm返回成功returns不触发消息丢失模拟

生产者代码:

import com.stages.payment.start.config.mq.RabbitMQConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;

@SpringBootTest
class StagesPaymentStartApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() throws IOException {
        // 成功或失败都会触发
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println(Thread.currentThread().getName()+"消息发送交换机结果: "+ack);
            }
        });
        // 发送到队列失败时才会触发
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                System.out.println(Thread.currentThread().getName()+"消息发送队列结果: "+returned);
            }
        });
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, "black.rabbit.male", "这里是消息4", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                messageProperties.setCorrelationId("123");
                messageProperties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);  // 这里设置非持久化
                return message;
            }
        });
        System.out.println(Thread.currentThread().getName()+"程序没有中断");
    }

}

运行结果:数据在内存中,rabbitmq宕机重启数据就丢失
【RabbitMQ Confirm和returns保证生产者消息可靠性】
【RabbitMQ Confirm和returns保证生产者消息可靠性】

结论

单独的returns也不可能保证消息的可靠性。

最后方案

使用 交换机持久化 + 队列持久化 + confirm + returns + 消息持久化 来保证消息的可靠性。

最终代码:

import com.stages.payment.start.config.mq.RabbitMQConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;

@SpringBootTest
class StagesPaymentStartApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() throws IOException {
        // message发送到交换机的回调补偿
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println(Thread.currentThread().getName()+"消息发送交换机结果: "+ack);
            }
        });
        // message发送到队列失败的补偿
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                System.out.println(Thread.currentThread().getName()+"消息发送队列结果: "+returned);
            }
        });
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, "black.rabbit.male", "这里是消息4", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                // 设置消息的持久化属性 默认为持久化的。
                messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                return message;
            }
        });
        System.out.println(Thread.currentThread().getName()+"程序没有中断");
    }

}

转载请注明出处!!!

w3cjava

发表评论

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen: