RabbitMq消息防丢失功能实现方式讲解

目录
  • 1.概述
    • 1.1.数据丢失的原因
    • 1.2.如何防止数据丢失
  • 2.手动应答
  • 3.消息确认机制
    • 3.1.AMQP事务
    • 3.2.confirm

1.概述

1.1.数据丢失的原因

在消息中有三种可能性造成数据丢失:

  • 消费者消费消息失败
  • 生产者生产消息失败
  • MQ数据丢失

消费者消费消息失败:

RabbitMq存在应答机制,默认为自动应答,MQ向消费者推送一条消息,消费者收到这条消息后会返回一个ack(应答)给MQ,MQ收到应答后会删除这条消息。

自动应答存在一个问题,就是消费者收到消息后立马就会给MQ返回ack,如果消费者返回完ack但还没来的及真正处理这条消息时,消费者断电宕机了,那么这条消息就丢失了。

这就是由于消费者消费消息失败造成的数据丢失。

生产者生产数据失败:

生产者向MQ推送了一条消息,但是由于由于诸如网络故障等原因mq并没有收到该条消息,这样就造成了这条消息的丢失。

MQ数据丢失:

MQ的数据是存在内存中的,诸如断电等原因可能会造成数据的丢失。

1.2.如何防止数据丢失

解决以上列举的数据丢失问题的办法有三种:

  • 手动应答
  • 消息确认机制
  • 持久化

手动应答:

RabbitMQ默认是自动应答,消费者收到消息后就会自动返回ack给MQ,可以将应答模式改为手动应答,在消费者一侧消息的消费动作完成后手动来返回ack给MQ,用来解决“消费者消费消息失败”问题。

消息确认机制:

当消息队列收到消息后,告知生产者,让生产者感知到自己生产的消息,消息队列已经接收到,用来解决“生产者生产消息失败”问题。消息确认机制有两种实现方式:

  • AMQP事务
  • confirm

持久化:

消息队列的消息持久化到磁盘上,用来解决“MQ数据丢失”问题。

2.手动应答

手动应答是通过设置channel来实现的,以下为一个完整代码示例。

配置类:

@Configuration
public class config {
    @Bean
    public Queue queue(){
        return new Queue("queue_01",false);
    }
}

生产者:

@SpringBootTest(classes = Main.class)
public class Producer {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    public void producerMsg(){
        rabbitTemplate.convertAndSend("queue_01","hello_world");
    }
}

消费者:

@Component
@Slf4j
public class Consumer {
    @RabbitListener(queues = {"queue_01"})
    public void consumerMsg(String msg, Message message,Channel channel){
        try {
            log.info("消费者消费消息: "+msg);
            /**
             * 没有异常就确认消息
             * basicAck(long deliveryTag, boolean multiple)
             * deliveryTag:当前消息在队列中的的索引;
             * multiple:为true的话就是批量确认
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            /**
             * 有异常就拒收消息
             * basicNack(long deliveryTag, boolean multiple, boolean requeue)
             * requeue:true为将消息重返当前消息队列,重新发送给消费者;
             *         false将消息丢弃
             */
            try {
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            } catch (Exception ex) {
                log.error(ex.getMessage());
            }
        }
    }
}

3.消息确认机制

AQMP事务、confirm其实都是基于channel的。

3.1.AMQP事务

AMQP事务和数据库事务类似,定义一组对MQ的操作,统一提交,成功则全部一起执行,失败则全部回滚。AMQP事务在spring boot中的使用很简单,和数据库的事务一样,一个注解就可以搞定。

@GetMapping("/direct/wx/transactional")
@Transactional(rollbackFor = Exception.class)
public String sendDirectMessageTransactional() {
  rabbitTemplate.convertAndSend("direct_exchange", "wx","hello world!");
  log.info("开启事务消息机制");
    try {
           Thread.sleep(5000);
       } catch (Exception e) {
            e.printStackTrace();
       }
      return "ok";
}

3.2.confirm

confirm是基于channel的,一旦channel进入confirm模式,所有在该channel上发布的消息都会被指派一个唯一的ID(从1开始),消息被投递道匹配队列后broker会发送一个确认消息给生产者。如果消息和队列是可持久化的(durable为true),那么确认消息会在消息被写入磁盘后发出。

confirm最大的好处在于异步,生产者在等待上一条消息的确认消息的时候可以继续往下发送。

confirm在spring boot中的使用很简单,在配置文件中开启即可,并且支持自定义回调函数:

配置文件:

spring.rabbitmq.publisher-confirms: true

spring.rabbitmq.publisher-returns: true

生产者:

@Slf4j
@Component
public class RabbitmqService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMessage(String exchange,String routingKey,Object msg) {
        // 设置交换机处理失败消息的模式     true 表示消息由交换机 到达不了队列时,会将消息重新返回给生产者
        // 如果不设置这个指令,则交换机向队列推送消息失败后,不会触发 setReturnCallback
        rabbitTemplate.setMandatory(true);
        //消息消费者确认收到消息后,手动ack回执
        rabbitTemplate.setConfirmCallback(this);
        // 暂时关闭 return 配置
        //rabbitTemplate.setReturnCallback(this);
        //发送消息
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }
    /**
     * 交换机并未将数据丢入指定的队列中时,触发
     *  channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes());
     *  参数三:true  表示如果消息无法正常投递,则return给生产者 ;false 表示直接丢弃
     * @param message   消息对象
     * @param replyCode 错误码
     * @param replyText 错误信息
     * @param exchange 交换机
     * @param routingKey 路由键
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" ");
    }
    /**
     * 消息生产者发送消息至交换机时触发,用于判断交换机是否成功收到消息
     * @param correlationData  相关配置信息
     * @param ack exchange 交换机,判断交换机是否成功收到消息    true 表示交换机收到
     * @param cause  失败原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("---- confirm ----ack="+ack+"  cause="+String.valueOf(cause));
        log.info("correlationData -->"+correlationData.toString());
        if(ack){
            // 交换机接收到
            log.info("---- confirm ----ack==true  cause="+cause);
        }else{
            // 没有接收到
            log.info("---- confirm ----ack==false  cause="+cause);
        }
    }
}

到此这篇关于RabbitMq消息防丢失功能实现方式讲解的文章就介绍到这了,更多相关RabbitMq消息防丢失内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • Spring boot Rabbitmq消息防丢失实践

    目录 前言 导致消息出现丢失的原因 环境 准备工作 使用confirm机制 模拟场景 实现RabbitTemplate.ConfirmCallback接口 发送端代码 实现效果 使用return机制 模拟场景 实现RabbitTemplate.ReturnCallback 发送端代码 实现效果 rabbitmq服务挂了,造成内存的消息丢失. 发送到消费端消费失败 修改application.yml配置文件 消费了,但是忘记做手动确认ack的操作代码. 效果 消费过程中,触发了未知异常,代码没有t

  • RabbitMq消息防丢失功能实现方式讲解

    目录 1.概述 1.1.数据丢失的原因 1.2.如何防止数据丢失 2.手动应答 3.消息确认机制 3.1.AMQP事务 3.2.confirm 1.概述 1.1.数据丢失的原因 在消息中有三种可能性造成数据丢失: 消费者消费消息失败 生产者生产消息失败 MQ数据丢失 消费者消费消息失败: RabbitMq存在应答机制,默认为自动应答,MQ向消费者推送一条消息,消费者收到这条消息后会返回一个ack(应答)给MQ,MQ收到应答后会删除这条消息. 自动应答存在一个问题,就是消费者收到消息后立马就会给M

  • Rabbitmq消息推送功能实现示例

    目录 一.前言 1.1场景 1.2消息交换机三种形式 二.建设demo工程 2.1依赖 2.2yml文件指定rabbitmq连接信息 2.3直连型消息链接 一.前言 1.1场景 在我们实际开发中到一个特定的时候是比如工作流到某个状态时, 我们会向某某单位发送消息, 这时就会用到我们的消息推送---rabbitmq 简单画一下: 1.2消息交换机三种形式 首先我们了解下消息队列是由交换机exchange和队列组合构成的,有三种形式 直连型:一个交换机关联一个队列,指定一个路由key,消息通过交换机

  • PHP实现长轮询消息实时推送功能代码实例讲解

    本文实例讲述了PHP实现的消息实时推送功能.分享给大家供大家参考,具体如下: 入口文件index.html <!DOCTYPE HTML> <html> <head> <title>反ajax推送</title> <style> .send{color:#555;text-align: left;} .require{color:blue;text-align: right;} .content_box{text-align: cen

  • Springcloud整合stream,rabbitmq实现消息驱动功能

    springcloud整合stream,rabbitmq实现消息驱动功能 1.代码实现: 创建项目stream 添加依赖 <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.6.2</version> <relativePath/>

  • 详解SpringBoot中使用RabbitMQ的RPC功能

    一.RabbitMQ的RPC简介 实际业务中,有的时候我们还需要等待消费者返回结果给我们,或者是说我们需要消费者上的一个功能.一个方法或是一个接口返回给我们相应的值,而往往大型的系统软件,生产者跟消费者之间都是相互独立的两个系统,部署在两个不同的电脑上,不能通过直接对象.方法的形式获取想要的结果,这时候我们就需要用到RPC(Remote Procedure Call)远程过程调用方式. RabbitMQ实现RPC的方式很简单,生产者发送一条带有标签(消息ID(correlation_id)+回调

  • RabbitMQ消息有效期与死信的处理过程

    目录 一.前言 二.设置消息有效期 1.设置队列的有效期TTL 2.设置队列的有效期Expire 3.通过发送消息时设置有效期 三.死信交换机DLX 一.前言 RabbitMQ的TTL全称为Time-To-Live,表示的是消息的有效期.消息如果在队列中一直没有被消费并且存在时间超过了TTL,消息就会变成了"死信" (Dead Message),后续无法再被消费了.如果不设置TTL,则表示此消息永久有效(默认消息是不会失效的).如果将TTL设为0,则表示如果消息不能被立马消费则会被立即

  • 利用Python学习RabbitMQ消息队列

    RabbitMQ可以当做一个消息代理,它的核心原理非常简单:即接收和发送消息,可以把它想象成一个邮局:我们把信件放入邮箱,邮递员就会把信件投递到你的收件人处,RabbitMQ就是一个邮箱.邮局.投递员功能综合体,整个过程就是:邮箱接收信件,邮局转发信件,投递员投递信件到达收件人处. RabbitMQ和邮局的主要区别就是RabbitMQ接收.存储和发送的是二进制数据----消息. rabbitmq基本管理命令: 一步启动Erlang node和Rabbit应用:sudo rabbitmq-serv

  • jQuery+ThinkPHP+Ajax实现即时消息提醒功能实例代码

    心血来潮想为自己的小项目做一个提醒系统,譬如私信,评论等消息都能及时传递过来.由于道行尚浅,网上那些长轮询对于我略微复杂,于是觉得还是自己写一写试试比较好. 我的思路是,单独在数据库中建一个提醒表,表主要由接收者的id和消息类型两个字段组成 /* 前台提醒表 */ CREATE TABLE IF NOT EXISTS notification( id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, -- 主键自增 mid INT NOT NULL DEFAULT

  • 微信推送功能实现方式图文详解

    推送的方式: 短信推送(第三方) 邮件推送 微信推送 公众号:认证的公众号(个人的认证公众号每天只能发一篇文章),粉丝可以跟公众号聊天,未认证公众号 服务号:企业认证(营业执照),沙箱环境测试主动给用户发消息(推送),用户要接收到推送消息前提是需要关注对应的服务号才行 企业号 微信小程序 微信推送的流程: 微信沙箱环境:http://mp.weixin.qq.com/debug/cgi-bin/sandbox?t=sandbox/login 使用pycharm打开微信推送demo,然后进行相关修

  • SpringBoot整合RabbitMQ消息队列的完整步骤

    SpringBoot整合RabbitMQ 主要实现RabbitMQ以下三种消息队列: 简单消息队列(演示direct模式) 基于RabbitMQ特性的延时消息队列 基于RabbitMQ相关插件的延时消息队列 公共资源 1. 引入pom依赖 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId>

随机推荐