RocketMQ消息丢失场景以及解决方法

既然使用在项目中使用了MQ,那么就不可避免的需要考虑消息丢失问题。在一些涉及到了金钱交易的场景下,消息丢失还是很致命的。那么在RocketMQ中存在哪几种消息丢失的场景呢?

先来一张最简单的消费流程图:

上图中大致包含了这么几种场景:

生产者产生消息发送给RocketMQRocketMQ接收到了消息之后,必然需要存到磁盘中,否则断电或宕机之后会造成数据的丢失消费者从RocketMQ中获取消息消费,消费成功之后,整个流程结束

这三种场景都可能会产生消息的丢失,如下图所示:

场景1中生产者将消息发送给Rocket MQ的时候,如果出现了网络抖动或者通信异常等问题,消息就有可能会丢失场景2中消息需要持久化到磁盘中,这时会有两种情况导致消息丢失

①RocketMQ为了减少磁盘的IO,会先将消息写入到os cache中,而不是直接写入到磁盘中,消费者从os cache中获取消息类似于直接从内存中获取消息,速度更快,过一段时间会由os线程异步的将消息刷入磁盘中,此时才算真正完成了消息的持久化。在这个过程中,如果消息还没有完成异步刷盘,RocketMQ中的Broker宕机的话,就会导致消息丢失

②如果消息已经被刷入了磁盘中,但是数据没有做任何备份,一旦磁盘损坏,那么消息也会丢失消费者成功从RocketMQ中获取到了消息,还没有将消息完全消费完的时候,就通知RocketMQ我已经将消息消费了,然后消费者宕机,但是RocketMQ认为消费者已经成功消费了数据,所以数据依旧丢失了

那么如何保证消息的零丢失呢?

1、场景1中保证消息不丢失的方案是使用RocketMQ自带的事务机制来发送消息,大致流程为

①首先生产者发送half消息到RocketMQ中,此时消费者是无法消费half消息的,若half消息就发送失败了,则执行相应的回滚逻辑

②half消息发送成功之后,且RocketMQ返回成功响应,则执行生产者的核心链路

③如果生产者自己的核心链路执行失败,则回滚,并通知RocketMQ删除half消息

④如果生产者的核心链路执行成功,则通知RocketMQ commit half消息,让消费者可以消费这条数据

其中还有一些RocketMQ长时间没有收到生产者是要commit/rollback操作的响应,回调生产者接口的细节,感兴趣的可以参考文末的 RocketMQ分布式事务原理

在使用了RocketMQ事务将生产者的消息成功发送给RocketMQ,就可以保证在这个阶段消息不会丢失在场景2中要保证消息不丢失,首先需要将os cache的异步刷盘策略改为同步刷盘,这一步需要修改Broker的配置文件,将flushDiskType改为SYNC_FLUSH同步刷盘策略,默认的是ASYNC_FLUSH异步刷盘。一旦同步刷盘返回成功,那么就一定保证消息已经持久化到磁盘中了;为了保证磁盘损坏不会丢失数据,我们需要对RocketMQ采用主从机构,集群部署,Leader中的数据在多个Follower中都存有备份,防止单点故障。在场景3中,消息到达了消费者,RocketMQ在代码中就能保证消息不会丢失

//注册消息监听器处理消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
 @Override
 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){
  //对消息进行处理
  return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
 }
});

上面这段代码中,RocketMQ在消费者中注册了一个监听器,当消费者获取到了消息,就会去回调这个监听器函数,去处理里面的消息

当你的消息处理完毕之后,才会返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS
只有返回了CONSUME_SUCCESS,消费者才会告诉RocketMQ我已经消费完了,此时如果消费者宕机,消息已经处理完了,也就不会丢失消息了

如果消费者还没有返回CONSUME_SUCCESS时就宕机了,那么RocketMQ就会认为你这个消费者节点挂掉了,会自动故障转移,将消息交给消费者组的其他消费者去消费这个消息,保证消息不会丢失

为了保证消息不会丢失,在consumeMessage方法中就直接写消息消费的业务逻辑就可以了,如果非要搞一些骚操作,比如下面的代码

//注册消息监听器处理消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
 @Override
 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){
 	//开启子线程异步处理消息
 	new Thread() {
			public void run() {
				//对消息进行处理
			}
		}.start();
  return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
 }
});

如果新开子线程异步处理消息的话,就有可能出现消息还没有被消费完,消费者告诉RocketMQ消息已经被消费了,结果宕机丢失消息的情况。

使用上面一整套的方案就可以在使用RocketMQ时保证消息零丢失,但是性能和吞吐量也将大幅下降

使用事务机制传输消息,会比普通的消息传输多出很多步骤,耗费性能同步刷盘相比异步刷盘,一个是存储在磁盘中,一个存储在内存中,速度完全不是一个数量级主从架构的话,需要Leader将数据同步给Follower消费时无法异步消费,只能等待消费完成再通知RocketMQ消费完成

消息零丢失是一把双刃剑,要想用好,还是要视具体的业务场景而定,选择合适的方案才是最好的

RocketMQ分布式事务原理

分布式事务常见的方案有TCC(Try-Confirm-Cancel),XA两阶段提交方案,可靠消息最终一致性方案,最大努力通知方案等等。其中可靠消息最终一致性方案主要就可以依靠RocketMQ来做,因为RocketMQ支持消息事务。先上一张图:

RocketMQ 事务消息的实现步骤如下:

  1. Producer发送half message给RocketMQ
  2. RocketMQ返回half message success(half message发送成功之后RocketMQ的消费者并不能消费这条消息,因为消息存储在Topic为 RMQ_SYS_TRANS_HALF_TOPIC 的消息消费队列中,而不是原先的Topic)
  3. 执行核心交易链路
  4. 返回执行交易链路的结果,如果失败则回滚
  5. 如果执行成功,则Producer返回一个COMMIT状态给RocketMQ
  6. 如果RocketMQ迟迟收不到Producer的返回结果,即这条消息的状态为UNKNOWN,则会回调服务接口,查询这条消息到底是commit还是rollback
  7. RocketMQ确认消息为commit,则Consumer可以消费到这条消息
  8. Consumer操作数据库,执行自己的事务
  9. Consumer成功消费之后返回一个ACK消息给RocketMQ,如果成功消费则显示消费成功,否则RocketMQ会重发消息给Consumer继续消费

RocketMQ 事务消息的实现原理基于两阶段提交和定时事务状态回查来决定消息最终是提交还是回滚,RocketMQ 先执行第一部分的事务,如果失败则回滚,如果成功则定时任务会去回查到事务执行成功,这个时候通知消费者执行第二阶段的事务,如果失败则不断重发消息给消费者消费,如果成功则整个流程走完,保证了事务的原子性。

总结

到此这篇关于RocketMQ消息丢失场景以及解决方法的文章就介绍到这了,更多相关RocketMQ消息丢失场景内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

时间: 2020-09-15

RocketMq事务消息发送代码流程详解

一.RocketMq事务消息流程: 1.首先会向broker发送一个预请求消息,消费者不可见 2.回调执行本地事务(比如操作数据库) 3.事务执行成功后,再次发送消息给broker,告诉broker事务执行成功这个消息要提交,让消费者可见.如果本地事务执行超时,会返回一个unknow,broker会发送一个消息回查,检查消息是否执行成功. 二.RocketMq事务消息实例: 1.引入rocketMq相关的依赖: <dependency> <groupId>org.apache.ro

使用Kotlin+RocketMQ实现延时消息的示例代码

一. 延时消息 延时消息是指消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者才拿到这个消息进行消费. 使用延时消息的典型场景,例如: 在电商系统中,用户下完订单30分钟内没支付,则订单可能会被取消. 在电商系统中,用户七天内没有评价商品,则默认好评. 这些场景对应的解决方案,包括: 轮询遍历数据库记录 JDK 的 DelayQueue ScheduledExecutorService 基于 Quartz 的定时任务 基于 Redis 的 zset 实现延时队列. 除此之外,

java rocketmq--消息的产生(普通消息)

前言 与消息发送紧密相关的几行代码: 1. DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); 2. producer.start(); 3. Message msg = new Message(...) 4. SendResult sendResult = producer.send(msg); 5. producer.shutdown(); 那这几行代码执行时,背后都做了什么? 一. 首先

RocketMQ重试机制及消息幂代码实例解析

这篇文章主要介绍了RocketMQ重试机制及消息幂代码实例解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 一.重试机制 1.由于MQ经常处于复杂的分布式系统中,考虑网络波动,服务宕机,程序异常因素,很有可能出现消息发送或者消费失败的问题.因此,消息的重试就是所有MQ中间件必须考虑到的一个关键点.如果没有消息重试,就可能产生消息丢失的问题,可能对系统产生很大的影响.所以,秉承宁可多发消息,也不可丢失消息的原则,大部分MQ都对消息重试提供了很好

RocketMQ获取指定消息的实现方法(源码)

概要 消息查询是什么? 消息查询就是根据用户提供的msgId从MQ中取出该消息 RocketMQ如果有多个节点如何查询? 问题:RocketMQ分布式结构中,数据分散在各个节点,即便是同一Topic的数据,也未必都在一个broker上.客户端怎么知道数据该去哪个节点上查? 猜想1:逐个访问broker节点查询数据 猜想2:有某种数据中心存在,该中心知道所有消息存储的位置,只要向该中心查询即可得到消息具体位置,进而取得消息内容 实际: 1.消息Id中含有消息所在的broker的地址信息(IP\Po

JavaScript获取指定元素位置的方法

本文实例讲述了JavaScript获取指定元素位置的方法.分享给大家供大家参考.具体如下: 复制代码 代码如下: function showpane() {   var self = document.getElementById("eID");   var left = self.getBoundingClientRect().left + document.documentElement.scrollLeft;   var top = self.getBoundingClientR

Java终止线程实例和stop()方法源码阅读

了解线程 概念 线程 是程序中的执行线程.Java 虚拟机允许应用程序并发地运行多个执行线程. 线程特点 拥有状态,表示线程的状态,同一时刻中,JVM中的某个线程只有一种状态; ·NEW 尚未启动的线程(程序运行开始至今一次未启动的线程) ·RUNNABLE 可运行的线程,正在JVM中运行,但它可能在等待其他资源,如CPU. ·BLOCKED 阻塞的线程,等待某个锁允许它继续运行 ·WAITING 无限等待(再次运行依赖于让它进入该状态的线程执行某个特定操作) ·TIMED_WAITING 定时

thinkphp3.2.0 setInc方法 源码全面解析

我们先来看一下setInc的官方示例: 需要一个字段和一个自增的值(默认为1) 我们通过下面这个例子来一步步分析他的底层是怎么实现的: <?php namespace Home\Controller; use Think\Controller; class TestController extends Controller { public function test() { $tb_test = M('test'); $tb_test->where(['id'=>1])->set

Spring SpringMVC在启动完成后执行方法源码解析

关键字:spring容器加载完毕做一件事情(利用ContextRefreshedEvent事件) 应用场景:很多时候我们想要在某个类加载完毕时干某件事情,但是使用了spring管理对象,我们这个类引用了其他类(可能是更复杂的关联),所以当我们去使用这个类做事情时发现包空指针错误,这是因为我们这个类有可能已经初始化完成,但是引用的其他类不一定初始化完成,所以发生了空指针错误,解决方案如下: 1.写一个类继承spring的ApplicationListener监听,并监控ContextRefresh

CentOS 6.3 安装配置Apache2.2.6的方法(源码编译安装)

安装说明 安装环境:CentOS-6.3 安装方式:源码编译安装 软件:httpd-2.2.6.tar.gz | pcre-8.32.tar.gz | apr-1.4.6.tar.gz | apr-util-1.5.1.tar.gz 下载地址:http://mirror.bjtu.edu.cn/apache/httpd/ http://apr.apache.org/download.cgi http://jaist.dl.sourceforge.net/project/pcre/pcre 安装位

Java 获取指定日期的实现方法总结

格式化日期 String-->Date 或者 Data-->String SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd"); Date date = sdf.parse("2009-11-04");//String-->Date String sdate = sdf.format(date );// Data-->String =========================

jQuery1.5.1 animate方法源码阅读

复制代码 代码如下: /*7536-7646*/ animate: function( prop, speed, easing, callback ) { if ( jQuery.isEmptyObject( prop ) ) { return this.each( optall.complete ); } //#7864行this.options.complete.call( this.elem )使得其可以不断的连续执行动画,比如$('selector').animate({prop1},s

Android开发获取手机内网IP地址与外网IP地址的详细方法与源码实例

在进行Android应用开发过程中,有时候会遇到获取当前Android设备所使用的网络IP地址的场景,有时候需要本地的网络IP地址,即局域网地址,更多的时候是需要当前网络的真实的对外IP地址,即真实的网络地址,如大数据分析时往往需要Android设备上传本地的外网地址.本文对各种IP地址的获取进行了总结. 首先用大家比较熟悉的电脑端局域网地址和外网地址的获取方式对比一下:(1).电脑端局域网地址获取方式,可以通过在终端命令行输入ipconfig进行查看,如下图IPv地址标识的就是本机的局域网地址

java使用websocket,并且获取HttpSession 源码分析(推荐)

一:本文使用范围 此文不仅仅局限于spring boot,普通的spring工程,甚至是servlet工程,都是一样的,只不过配置一些监听器的方法不同而已. 本文经过作者实践,确认完美运行. 二:Spring boot使用websocket 2.1:依赖包 websocket本身是servlet容器所提供的服务,所以需要在web容器中运行,像我们所使用的tomcat,当然,spring boot中已经内嵌了tomcat. websocket遵循了javaee规范,所以需要引入javaee的包 <

Spring MVC 中获取session的几种方法(小结)

Spring MVC 中使用session是一种常见的操作,但是大家上网搜索一下可以看到获取session的方式方法五花八门,最近,自己终结了一下,将获取session的方法记录下来,以便大家共同学习进步. 第一种:将HttpSession作为Spring MVC 的方法参数传入,直接获取. 直接在Spring MVC 的方法中将参数传入: public void getSessionAction(HttpSession session){ } 这种方法我再网上搜索时发现很多人并不推荐使用,但是