RabbitMq 消息确认和退回机制

🕗 发布于 2024-07-03 20:35 rabbitmq 分布式

一、Rabbit中消息确认和退回机制

1、发布确认

生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID (从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认给生产者 (包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了，（单个）如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号。

1.1 开启发布确认

//开启发布确认
channel.confirmSelect();

1.2 发布确认的两种方式

同步确认

就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布waitForConfirmsOrDie(long) 这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。

public static void publishMessageIndividually() throws Exception {
    Channel channel = RabbitMqUtils.getChannel();
    //队列声明
    String queueName = UUID.randomUUID().toString();
    channel.queueDeclare(queueName, true, false, false, null);
    //开启发布确认
    channel.confirmSelect();

    long begin = System.currentTimeMillis();

    for (int i = 0; i < MESSAGE_COUNT; i++) {
        String message = i + "";
        channel.basicPublish("", queueName, null, message.getBytes());
        //服务端返回 false 或超时时间内未返回，生产者可以消息重发
        boolean flag = channel.waitForConfirms();
        if (flag) {
            System.out.println("消息发送成功");
        }
    }

    long end = System.currentTimeMillis();
    System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) + "ms");

}

异步确认

他是利用回调函数来达到消息可靠性传递的

//    异步确认发布
    private static void publishMessageAsync() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();


        /**线程安全有序的一个哈希表，适用于高并发的情况下
          1.轻松的将序号与消息进行关联
          2.轻松批量删除条目，只要给到序号
          3.支持高并发（多线程）
         **/
        //消息确认成功，回调函数
        ConcurrentSkipListMap<Long,String> outstandingConfirms =
                new ConcurrentSkipListMap<>();
        ConfirmCallback ackCallback = (deliveryTag , multiple)->{ //确认了多少条，multiple:批量或单个
            if (multiple){  //批量的
                //2.轻松批量删除条目，只要给到序号
                ConcurrentNavigableMap<Long,String> confirmed =
                        outstandingConfirms.headMap(deliveryTag); //消息的序号
                confirmed.clear();
            }else {  //单个的
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("确认的消息："+deliveryTag);
        };

        //消息确认失败，回调函数
        /**
         * 1.消息的标记
         * 2.是否为批量确认
         */
        ConfirmCallback nackCallback = (deliveryTag , multiple)->{
            //3、打印一下未确认的消息都有哪些
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("未确认的消息是:"+message+":::未确认的消息："+deliveryTag);
        };


        //准备消息的监听器 ，监听哪些消息成功了， 哪些失败了
        channel.addConfirmListener( ackCallback, nackCallback);

        //开始时间
        long begin = System.currentTimeMillis();

        //发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            //1.轻松的将序号与消息进行关联，将每一条消息都存放hash表里面
            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
        }

        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) + "ms");
    }

2、退回机制

退回模式（return）说的是当消息到达交换机后，但是没有找到匹配的队列时，将消息回退给生产者。

默认情况下，如果消息没有匹配到队列会直接丢弃，采用退回模式可以在生产者端监听改消息是否被成功投递到队列中

channel.addReturnListener(new ReturnCallback() {
                @Override
                public void handle(Return returnMessage) {
                    System.out.println("消息被回退，原因："+returnMessage.getReplyText());
                    System.out.println(returnMessage.getExchange()); // 交换机
                    System.out.println(returnMessage.getReplyCode()); // 返回原因的代码
                    System.out.println(returnMessage.getReplyText()); // 返回信息，例如NO_ROUTE
                    System.out.println(returnMessage.getRoutingKey()); // 路由KEY
                }
            });

二、Spring boot 中Rabbit 实现消息确认和退回机制

1、开启配置

spring:
  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虚拟host 可以不设置,使用server默认host
    virtual-host: JCcccHost
      #确认消息已发送到交换机(Exchange)
    publisher-confirm-type: correlated
      #确认消息已发送到队列(Queue)
    publisher-returns: true

2、配置消息确认和退回机制

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Configuration
public class RabbitConfig {
 
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
 
        // 配置发布到交换机的确认
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
            *correlationData :客户端在发送原始消息时提供的对象。
            *ack：exchange交换机是否成功收到了消息。true成功，false代表失败。
            *cause：失败原因。
            */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback:     "+"相关数据："+correlationData);
                System.out.println("ConfirmCallback:     "+"确认情况："+ack);
                System.out.println("ConfirmCallback:     "+"原因："+cause);
            }
        });
 
        // 配置交换机没有找到对应的消息队列时，消息退回时的处理
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback:     "+"消息："+message);
                System.out.println("ReturnCallback:     "+"回应码："+replyCode);
                System.out.println("ReturnCallback:     "+"回应信息："+replyText);
                System.out.println("ReturnCallback:     "+"交换机："+exchange);
                System.out.println("ReturnCallback:     "+"路由键："+routingKey);
            }
        });
 
        return rabbitTemplate;
    }
 
}

2.1 其他方式

package com.student.rabbitmq.config;


import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * 第二种方式
 */
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 将我们实现的MyCallBack接口注入到RabbitTemplate中
    @PostConstruct
    public void init() {
        // 设置确认消息交给谁处理
        rabbitTemplate.setConfirmCallback(this);
        // 设置回退消息交给谁处理
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * 交换机确认回调方法
     *
     * @param correlationData 保存回调消息的ID以及相关信息
     * @param ack             表示交换机是否收到消息(true表示收到)
     * @param cause           表示消息接收失败的原因(收到消息为null)
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            System.out.println("交换机已经收到ID为:{}的消息");
        } else {
            System.out.println("交换机还未收到ID为:{}的消息,原因为:{}" + cause);
        }
    }

    /**
     * 路由出现问题的消息回退方法
     *
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println(new String(message.getBody())+
                        exchange+
                        replyText+
                        routingKey);
    }

}

三、参考

发布确认—发布确认逻辑和发布确认的策略

消息可靠性之发布确认、退回机制_rabbittemplate

原文地址：https://blog.csdn.net/szchen2/article/details/140126019

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：路由器是什么？
下一篇：William Yang：从区块链先锋到艺术平台创始人

MySQL数据库基础
在过去的mysql中主要是使用的utf8mb3，但是mysql显示的就是utf8，中mysql8中区分显示了。类似于不同的编译器，虽然C/C++的代码是一样的，但是预处理、编译、汇编、链接这些过程却不
阅读更多2024-09-30
Percona Monitoring and Management
Percona Monitoring and Management (PMM)是一款开源的专用于管理和监控MySQL、MongoDB、PostgreSQL。
阅读更多2024-09-30
鸿蒙开发（NEXT/API 12）【已连接穿戴设备查询】手机侧应用开发
Wear Engine提供查询用户已连接的穿戴设备列表（即支持Wear Engine能力且与手机侧运动健康App处于连接状态的穿戴设备）的接口。
阅读更多2024-09-30
Redis数据库
本篇文章介绍了redis的相关知识，并使用springboot集成redis，实现缓存和分布式锁，后续会对文章勘误及更新~~
阅读更多2024-09-30
旧版的存档
【代码】旧版的存档。
阅读更多2024-09-30
Spring Boot 封装统一返回结果及全局异常处理
为了更细致地控制异常处理，我们可以定义一些自定义异常。然后，在全局异常处理器中添加对自定义异常的处理。通过封装统一的返回结果和全局异常处理，我们可以让Spring Boot应用更加健壮和易于维护。这种
阅读更多2024-09-30
【C++】IO流
C++IO流，包含输入输出流、文件流、字符流
阅读更多2024-09-30
低至1元/小时：国庆七天，30元通关《黑神话：悟空》！
随着《黑神话：悟空》自8月20日全球同步上线，正式登陆PC、PS5平台以来，以其精湛的画面和流畅的战斗体验，在发售三天后，该作的全平台销量超过1000万套，打破中国游戏历史记录，被媒体称为“中国首款3
阅读更多2024-09-30
阿里巴巴国际站获取商品详情item_get接口技术分享
item_get API接口是阿里巴巴开放平台提供的一个重要接口，它允许商家通过API调用，获取阿里巴巴平台上的商品详细信息。这些信息包括商品标题、价格、库存、属性、描述等，为商家提供了全面、准确的商
阅读更多2024-09-30
软件测试谣言二三事，认真你就输了
软件测试的职业寿命，取决于互联网行业能存活多久，至少目前看来，这个职业没有消失的风险，至于你能在这个职业待多久，那取决于你自己的能力，我见过不少超过35岁的老员工还在测试的职位上兢兢业业。以广州为例，
阅读更多2024-09-30