Spring boot Rabbitmq消息防丢失实践

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

前言

之前看很多网上大佬的防丢失的文章,文章中理论知识偏多,所以自己想着实践一下,实践过程中也踩了一些坑,因此写出了这篇文章。如果文章有误人子弟的地方,望在评论区指出。

导致消息出现丢失的原因

  • 发送时失败,指发送端发送完消息准备到达消息队列的过程中,因网络波动、消息队列服务宕机等,消息队列服务无法接收消息,所以导致了丢失。
  • 到达时宕机,消息队列服务接收到消息之后,如果没有开启持久化,消息会存储在内存中(当然内存吃紧的话,也会转入磁盘,缓解内存),如果这个时候服务挂了,那么内存中的消息就会丢失。
  • 发送到消费端失败,消费端接收到了消息的时候,消费端服务挂了,而rabbitmq默认自动ack,也就是说rabbitmq发送到消费端,一旦认定了消费端接收了,无论有无消费成功,rabbitmq都认为是发送成功。

下面我们以这三种情况进行实践。

环境

jdk1.8
Spring boot 2.3.7.RELEASE
Spring-boot-starter-amqp 2.3.7.RELEASE
Rabbitmq 3.7.7

准备工作

我事先准备了好了交换机以及队列:

  • 交换机:message.log.test.exchangemessage.log.test2.exchange
  • 队列:message.loss.test.queue

其中message.loss.test.queuemessage.log.test.exchange是绑定关系,而message.log.test2.exchange没有绑定队列

1.发送时失败

发送时失败,rabbitmq有两种情况是属于发送时失败。

  • 消息未到rabbitmq的交换机(exchange)
  • 消息到达了rabbitmq的交换机(exchange),但是没有到达队列(queue)

第一种的解决方式是使用confirm机制。第二种解决方式则是使用return机制

使用confirm机制

模拟场景

confirm机制是当发送端的消息没有到达rabbitmq的交换机(exchange)时,会触发confirm方法,告诉发送端该消息没有到达rabbitmq,需要做业务处理。
这里我们发送消息到rabbitmq不存在的交换机上,就可以模拟上述场景。

实现RabbitTemplate.ConfirmCallback接口

/**
 * 当消息没有到达Rabbitmq的交换机时触发该方法(当然到达了也会触发,)
 */
@Component
public class ConfirmCallBack implements RabbitTemplate.ConfirmCallback {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){

        rabbitTemplate.setConfirmCallback(this);
    }

    /**
     *
     * @param correlationData 消息属性体
     * @param ack 是否成功,成功到达true,没有到达,false
     * @param cause rabbitmq自身给的信息
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        //第一个坑,如果发送端发送消息时没有对correlationData进行处理,conirm方法接收到的对象都会是null
        //当接收失败并且correlationData对象为null,证明目前已经无法追溯回业务,可以做业务日志处理
        if(!ack&&correlationData==null){
            System.out.println(cause);
            //日志处理。。。

            return;
        }
        //如果接收失败
        if(!ack){
            System.out.println("消息Id:"+correlationData.getId());
            Message message=correlationData.getReturnedMessage();
            System.out.println("消息体:"+new String(message.getBody()));
            //这里可以持久化业务消息体到数据库,然后定时去进行补偿处理或者重试等等
            return;
        }

        //处理完成

    }
}

发送端代码

/**
 * 消息的推送
 * @return
 */
@PostMapping("push")
public boolean push(){

    TestMessage testMessage=new TestMessage();
    testMessage.setName("mq名称");
    testMessage.setBusinessId("业务Id");

    //定义CorrelationData对象以及消息属性。不然comfirm方法无论失败还是成功,CorrelationData参数永远是null
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    //传递业务数据
    correlationData.setReturnedMessage(new Message(JSONObject.toJSON(testMessage).toString().getBytes(StandardCharsets.UTF_8),new MessageProperties()));

  //发送消息(这里发送给了message.log.test.exchange11交换机,但实际rabbitmq并不存在)template.convertAndSend("message.log.test.exchange11","message_loss_test",testMessage,correlationData);

    return true;
}

这里是我踩的第一个坑,如果发送端不定义correlationData,那么confirm接收到的correlationData对象参数 都会是null

实现效果

使用return机制

模拟场景

当消息到达了rabbitmq的交换机的时候,但是又没有到达队列,那么就会触发return方法。
下面我们定义一个没有绑定队列的交换机,然后发送消息到交换机,就可以模拟上述场景

实现RabbitTemplate.ReturnCallback

/**
 * 当消息没有到达Rabbitmq的队列时就会触发该方法
 */
@Component
public class ReturnCallBack implements RabbitTemplate.ReturnCallback {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @PostConstruct
    public void init() {
        rabbitTemplate.setReturnCallback(this);
    }
    /**
     * @param message    消息体
     * @param replyCode  返回代码
     * @param replyText  返回文本
     * @param exchange   交换机
     * @param routingKey 发送方定义的路由key
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("消息标识:" + message.getMessageProperties().getDeliveryTag());
        String messageBody = null;
        try {
            messageBody = new String(message.getBody(), "UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        System.out.println("消息:" + messageBody);
        System.out.println(replyCode);
        System.out.println(replyText);
        System.out.println(exchange);
        System.out.println(routingKey);

    }
}

发送端代码

/**
 * 消息的推送
 * @return
 */
@PostMapping("push2")
public boolean push2(){

    TestMessage testMessage=new TestMessage();
    testMessage.setName("mq名称2");
    testMessage.setBusinessId("业务Id");

    template.convertAndSend("message.log.test2.exchange","message_loss_test",JSONObject.toJSON(testMessage).toString());

    return true;
}

这里需注意消息体需要JSON序列化,不然returnedMessage方法接收的消息body会是乱码

实现效果

rabbitmq服务挂了,造成内存的消息丢失。

这个开启rabbitmq的持久化机制就好了,开启之后消息到达rabbitmq服务,会实时转入磁盘。这里怎么设置就不多说了,网上挺多文章可以解答。

不过即使开启了还是会有一种情况会造成消息丢失,那就是消息即将要持久化到磁盘的那一刻,服务挂了,就会造成丢失,不过这种情况我也不知道怎么模拟,所以就暂不实践了。

发送到消费端消费失败

上面提到默认情况下rabbitmq使用的是自动ack的方式,我们将它改成手动ack的方式,就可以解决这个问题。

修改application.yml配置文件

rabbitmq:
 listener:
  simple:
    #开启手动确认
    acknowledge-mode: manual
    #开启失败后的重试机制
    retry:
      enabled: true
      #最多重试3次
      max-attempts: 3

下面我们试一下几种消费端消费不成功的场景

消费了,但是忘记做手动确认ack的操作代码。

@Component
public class TestConsumer {

    /**
     * 消费
     * @param testmessage 消息体
     * @param message 消息属性
     * @param channel mq通道对象
     */
    @RabbitListener(queues = {"message.loss.test.queue"})
    public void test(TestMessage testmessage, Message message, Channel channel) throws IOException {
        System.out.println("消费testmessage消息:"+testmessage.getName());
//        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

    }
}

效果

效果流程:

  • 第一次用Postman请求之后,控制台显示了消息被消费的信号。
  • 然后去查看rabbitmq后台管理刚刚被消费的消息以及变为Unacked
  • 停止程序后(关闭消费端),过一阵子,后台管理显示消息变回了Ready,也就是说重新回到了队列。
  • 重新启动程序(开启消费段),消息被重新消费。

总而言之,如果消费端没有做手动确认的操作,那么在消费端还没关闭之前,消息会变成Unacked,不会再次被消费,但一旦消费端关闭了,消息会重新回到队列,让消费端消费。

消费过程中,触发了未知异常,代码没有try catch

/**
 * 消费
 * @param testmessage 消息体
 * @param message 消息属性
 * @param channel mq通道对象
 */
@RabbitListener(queues = {"message.loss.test.queue"})
public void test(TestMessage testmessage, Message message, Channel channel) throws IOException {
    System.out.println("消费testmessage消息:"+testmessage.getName());
    //故意触发异常
    if(!StringUtils.isEmpty(testmessage.getName())){

        throw new RuntimeException("11211");
    }
    channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}

效果1

上面的效果图显示,我在触发了异常之后,消息重试了三次,也就是我在application.yml 配置的重试三次

如果我去掉重试机制会是什么效果。

效果2

效果和忘记做ack操作的效果一样,消息没有ack后,消息会变成Unacked状态,消费端关闭后消息会重新回到队列,然后重新链接的时候,就会再消费一次。

总结

到此这篇关于Spring boot Rabbitmq消息防丢失实践的文章就介绍到这了,更多相关Spring boot Rabbitmq 内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

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

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

  • Spring Boot+RabbitMQ 通过fanout模式实现消息接收功能(支持消费者多实例部署)

    本文章适用的场景:同一条消息可以被多个消费者同时消费.注意:当消费者多实例部署时,会轮询消费消息.网上有大量的的案例展示:P生产一条消息,消费者服务C中建立Q1和Q2两个队列共同消费.但极少的材料展示:P生产一条消息后M1,消费者C1和C2可以同时消费M1,如下图所示.案例基于Spring Boot以及RabbitMQ的“fanout”类型exchange.已经实测可放心使用. 1.引入基本依赖,项目不同请您按自己的情况引入合适的依赖 <dependency> <groupId>o

  • Spring Boot RabbitMQ 延迟消息实现完整版示例

    概述 曾经去网易面试的时候,面试官问了我一个问题,说 下完订单后,如果用户未支付,需要取消订单,可以怎么做 我当时的回答是,用定时任务扫描DB表即可.面试官不是很满意,提出: 用定时任务无法做到准实时通知,有没有其他办法? 我当时的回答是: 可以用队列,订单下完后,发送一个消息到队列里,并指定过期时间,时间一到,执行回调接口. 面试官听完后,就不再问了.其实我当时的思路是对的,只不过讲的不是很专业而已.专业说法是利用 延迟消息 . 其实用定时任务,确实有点问题,原本业务系统希望10分钟后,如果订

  • Spring Boot统一异常处理最佳实践(拓展篇)

    前言 之前一篇文章介绍了基本的统一异常处理思路: Spring MVC/Boot 统一异常处理最佳实践. 上篇文章也有许多人提出了一些问题: 如何区分 Ajax 请求和普通页面请求, 以分别返回 JSON 错误信息和错误页面. 如何结合 HTTP 状态码进行统一异常处理. 今天这篇文章就主要来讲讲这些, 以及其他的一些拓展点. 区分请求方式 其实 Spring Boot 本身是内置了一个异常处理机制的, 会判断请求头的参数来区分要返回 JSON 数据还是错误页面. 源码为: org.spring

  • 详解Guava Cache本地缓存在Spring Boot应用中的实践

    概述 在如今高并发的互联网应用中,缓存的地位举足轻重,对提升程序性能帮助不小.而 3.x开始的 Spring也引入了对 Cache的支持,那对于如今发展得如火如荼的 Spring Boot来说自然也是支持缓存特性的.当然 Spring Boot默认使用的是 SimpleCacheConfiguration,即使用 ConcurrentMapCacheManager 来实现的缓存.但本文将讲述如何将 Guava Cache缓存应用到 Spring Boot应用中. Guava Cache是一个全内

  • Spring Boot 单元测试JUnit的实践

    一.介绍 JUnit是一款优秀的开源Java单元测试框架,也是目前使用率最高最流行的测试框架,开发工具Eclipse和IDEA对JUnit都有很好的支持,JUnit主要用于白盒测试和回归测试. <!--more--> 白盒测试:把测试对象看作一个打开的盒子,程序内部的逻辑结构和其他信息对测试人 员是公开的: 回归测试:软件或环境修复或更正后的再测试: 单元测试:最小粒度的测试,以测试某个功能或代码块.一般由程序员来做,因为它需要知道内部程序设计和编码的细节: JUnit GitHub地址:ht

  • Spring Boot与Docker部署实践

    首先需要开启docker远程访问功能,以便可以进行远程操作. CentOS 6 修改/etc/default/docker文件,重启后生效(service docker restart). DOCKER_OPTS="-H=unix:///var/run/docker.sock -H=0.0.0.0:2375" CentOS 7 打开/usr/lib/systemd/system/docker.service文件,修改ExecStart这行. 复制代码 代码如下: ExecStart=/

  • 详解在spring boot中消息推送系统设计与实现

    推送系统作为通用的组件,存在的价值主要有以下几点 会被多个业务项目使用,推送系统独立维护可降低维护成本 推送系统一般都是调用三方api进行推送,三方api一般会有调用频率/次数限制,被推送的消息需要走队列来合理调用三方api,控制调用的频率和次数 业务无关,一般推送系统设计成不需要关心业务逻辑 核心技术 消息队列 三方服务api调用 安卓app推送 苹果app推送 微信小程序推送 邮件推送 钉钉推送 短信推送 消息队列选用阿里云提供的rocketmq,官方文档:https://help.aliy

  • spring boot使用RabbitMQ实现topic 主题

    前一篇我们实现了消息系统的灵活配置.代替了使用扇形(fanout)交换器的配置.使用直连(direct)交换器,并且基于路由键后可以有选择性接收消息的能力. 虽然使用直连交换器可以改善我们的系统,但是它仍有局限性,它不能实现多重条件的路由. 在我们的消息系统中,我们不仅想要订阅基于路由键的队列,还想订阅基于生产消息的源.这些概念来自于Unix工具syslog.该日志基于严格的(info/warn/crit...) 和容易的(auth/cron/kern...)的路由方式.我们的例子比这个要简单.

  • spring boot中使用RabbitMQ routing路由详解

    在上一个教程中我们创建了一个扇形(fanout)交换器.我们能把消息已广播的形式传递给多个消费者. 要做什么?Routing 路由 在这个教程中,添加一个新的特性,我们可以只订阅消息的一部分.例如,将只连接我们感兴趣的颜色("orange", "black", "green"),并且把消息全部打印在控制台上. 绑定 交换器和队列是一种绑定关系.简单的理解为:队列对来自这个交换器中的信息感兴趣. 绑定可以加上一个额外的参数routingKey.Sp

随机推荐