IT数码 购物 网址 头条 软件 日历 阅读 图书馆
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		图片批量下载器
↓批量下载图片,美女图库↓ 		图片自动播放器
↓图片自动播放器↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁
 
   -> Java知识库 -> 【RabbitMQ Confirm和returns保证生产者消息可靠性】 -> 正文阅读

[Java知识库]【RabbitMQ Confirm和returns保证生产者消息可靠性】

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

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控制台

队列为空。
在这里插入图片描述

创建交换机和队列并绑定

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根据消息属性决定存放在磁盘还是内存

在这里插入图片描述

简单理解两种机制

上图我们将消息发送简单分为了三个步骤。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()+"程序没有中断");
    }

}

运行结果：数据丢失且程序无报错。
在这里插入图片描述
在这里插入图片描述

confirm返回失败的情况模拟

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

运行结果：confirm返回失败且数据丢失
在这里插入图片描述
在这里插入图片描述

结论

单凭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触发
在这里插入图片描述
在这里插入图片描述

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宕机重启数据就丢失
在这里插入图片描述
在这里插入图片描述

结论

单独的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()+"程序没有中断");
    }

}

转载请注明出处！！！
  Java知识库 最新文章
计算距离春节还有多长时间
系统开发系列 之WebService（spring框架+ma
springBoot+Cache(自定义有效时间配置)
SpringBoot整合mybatis实现增删改查、分页查
spring教程
SpringBoot+Vue实现美食交流网站的设计与实
虚拟机内存结构以及虚拟机中销毁和新建对象
SpringMVC---原理
小李同学: Java如何按多个字段分组
打印票据--java
上一篇文章      下一篇文章      查看所有文章
加:2022-05-24 17:59:01  更:2022-05-24 18:03:07 
 
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁

360图书馆 购物 三丰科技 阅读网 日历 万年历 2024年11日历 -2024/11/23 20:29:43-

图片自动播放器
↓图片自动播放器↓ 		TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓ 		图片批量下载器
↓批量下载图片,美女图库↓
  网站联系: qq:121756557 email:121756557@qq.com  IT数码