RocketMq重试及消息不丢失机制

1、消息重试机制

由于MQ经常处于复杂的分布式系统中,考虑网络波动、服务宕机、程序异常因素,很有可能出现消息发送或者消费失败的问题。因此,消息的重试就是所有MQ中间件必须考虑到的一个关键点。如果没有消息重试,就可能产生消息丢失的问题,可能对系统产生很大的影响。所以,秉承宁可多发消息,也不可丢失消息的原则,大部分MQ都对消息重试提供了很好的支持。

RocketMQ为使用者封装了消息重试的处理流程,无需开发人员手动处理。RocketMQ支持了生产端和消费端两类重试机制。

1.1 生产端重试

生产端配置的有发送失败重试次数,默认为2。使用了set方法对外进行暴露,producer客户端可以改写这个默认值。

public DefaultMQProducer(String producerGroup, RPCHook rpcHook) {this.createTopicKey = "TBW102";this.defaultTopicQueueNums = 4;this.sendMsgTimeout = 3000;this.compressMsgBodyOverHowmuch = 4096;//发送失败,重试次数this.retryTimesWhenSendFailed = 2;this.retryAnotherBrokerWhenNotStoreOK = false;this.maxMessageSize = 131072;this.unitMode = false;this.producerGroup = producerGroup;this.defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);}

1.2 消费端重试

消费者消费消息后,需要给Broker返回消费状态。以MessageListenerConcurrently监听器为例,Consumer消费完成后需要返回ConsumeConcurrentlyStatus并发消费状态。查看源码,ConsumeConcurrentlyStatus是一个枚举,共有两种状态:

public enum ConsumeConcurrentlyStatus {//消费成功ConsumeConcurrentlyStatus,//消费失败,一段时间后重试RECONSUME_LATER;
}

RECONSUME_LATER代表因为某种原因,消费失败,稍后再试。后续会再次消费
官方文档介绍如下:
在这里插入图片描述
RocketMQ中的消息无法无限次重新消费,当然了,手动修改重试次数是可以的,不介入的话不行。当重试次数超过所有延迟级别之后。消息会进入死信,死信Topic的命名为:%DLQ% + Consumer组名。

进入死信之后的消息肯定不会再投递了,不过可以通过接口去查询当前RocketMQ中死信队列的消息。如果在上层实现自有命令,那么可以将消息从死信中移出并重新投递。

死信消息具有以下特性:

  • 不会再被消费者正常消费。
  • 有效期与正常消息相同,均为 3 天,3 天后会被自动删除。因此,请在死信消息产生后的 3 天内及时处理。

2、保证消息不丢失

分别从Producer发送机制、Broker的持久化机制,以及消费者的offSet机制来最大程度保证消息不易丢失

  • 从Producer的视角来看:如果消息未能正确的存储在MQ中,或者消费者未能正确的消费到这条消息,都是消息丢失。
  • 从Broker的视角来看:如果消息已经存在Broker里面了,如何保证不会丢失呢(宕机、磁盘崩溃)
  • 从Consumer的视角来看:如果消息已经完成持久化了,但是Consumer取了,但是未消费成功且没有反馈,就是消息丢失

从Producer分析:如何确保消息正确的发送到了Broker?

  • 默认情况下,可以通过同步的方式阻塞式的发送,check SendStatus,状态是OK,表示消息一定成功的投递到了Broker,状态超时或者失败,则会触发默认的2次重试。此方法的发送结果,可能Broker存储成功了,也可能没成功
  • 采取事务消息的投递方式,并不能保证消息100%投递成功到了Broker,但是如果消息发送Ack失败的话,此消息会存储在CommitLog当中,但是对ConsumerQueue是不可见的。可以在日志中查看到这条异常的消息,严格意义上来讲,也并没有完全丢失
  • RocketMQ支持 日志的索引,如果一条消息发送之后超时,也可以通过查询日志的API,来check是否在Broker存储成功

从Broker分析:如果确保接收到的消息不会丢失?

  • 消息支持持久化到Commitlog里面,即使宕机后重启,未消费的消息也是可以加载出来的
  • Broker自身支持同步刷盘、异步刷盘的策略,可以保证接收到的消息一定存储在本地的内存中
  • Broker集群支持 1主N从的策略,支持同步复制和异步复制的方式,同步复制可以保证即使Master 磁盘崩溃,消息仍然不会丢失

从Cunmser分析:如何确保拉取到的消息被成功消费?

  • 消费者可以根据自身的策略批量Pull消息
  • Consumer自身维护一个持久化的offset(对应MessageQueue里面的min offset),标记已经成功消费或者已经成功发回到broker的消息下标
  • 如果Consumer消费失败,那么它会把这个消息发回给Broker,发回成功后,再更新自己的offset
  • 如果Consumer消费失败,发回给broker时,broker挂掉了,那么Consumer会定时重试这个操作
  • 如果Consumer和broker一起挂了,消息也不会丢失,因为consumer 里面的offset是定时持久化的,重启之后,继续拉取offset之前的消息到本地

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/313452.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

cmake编译opencv3.0

本文参照了 http://www.huqiwen.com/2012/11/27/compile-opencv-243-in-visual-studio-2012/ 安装CMake 从CMake的官方网站下载最新版的CMake。http://www.cmake.org/cmake/resources/software.html,选择Windows (Win32 Installer)平台的进行下载。 安装时请勾选…

【 .NET Core 3.0 】框架之五 || JWT权限验证

前言关于JWT一共三篇 姊妹篇,内容分别从简单到复杂,一定要多看多想:一、Swagger的使用 3.3 JWT权限验证【修改】二、解决JWT权限验证过期问题三、JWT完美实现权限与接口的动态分配这里一共三个文章,目前是第一篇,剩下两…

OpenCV Stitching 工程搭建

转自http://www.tuicool.com/articles/fMbUfaF Opencv中提供Stitcher类,实现了多图像自动拼接,Opencv是开源的,程序实现的源代码都在Opencv安装文件中,以及Opencv提供的函数查询手册和Opencv教程都可以在…

asp.net core 3.0 更新简记

asp.net core 3.0 更新简记Intro最近把活动室预约项目从 asp.net core 2.2 更新到了 asp.net core 3.0,记录一下,升级踩过的坑以及经验总结,包括但不限于TargetFramework ( netcoreapp2.2 需要更新为 netcoreapp3.0)DependencyHost/Environme…

kafka吞吐量高的原因

kafa 吞吐量高的原因 1、顺序读写 kafka的消息是不断追加到文件中的,这个特性使kafka可以充分利用磁盘的顺序读写性能 顺序读写不需要硬盘磁头的寻道时间,只需很少的扇区旋转时间,所以速度远快于随机读写 2、零拷贝 在Linux kernel2.2 之…

【 .NET Core 3.0 】框架之三 || swagger的使用

一、为什么使用Swagger上文中已经说到,单纯的项目接口在前后端开发人员使用是特别不舒服的,那所有要推荐一个,既方便又美观的接口文档说明框架,当当当,就是Swagger,随着互联网技术的发展,现在的…

MQ问题集(kafka主从同步与高可用,MQ重复消费、幂等)

1、kafka主从同步与高可用 https://1028826685.iteye.com/blog/2354570 http://developer.51cto.com/art/201808/581538.htm 2、MQ有可能发生重复消费,如何避免,如何做到幂等 2.1 MQ消息发送 1、发送端MQ-client(消息生产者:Producer)将消…

微软编程题:寻找最小的k个值

转载自:http://blog.csdn.net/v_JULY_v/article/details/6370650 寻找最小的k个数 题目描述:5.查找最小的k个元素 题目:输入n个整数,输出其中最小的k个。 例如输入1,2,3,4,5&#xf…

Bumblebee微服务网关之访问日志处理

记录访问日志可以起到非常重要的作用,它不仅记录了API的使用情况,更可以反映API各种相关数据;通过分析日志可以得到API不同时间的负载情况,访问效率和流量分布,更进一步还能分析出用户的操作历史和行为这是非常有价值的…

负载均衡基础

1、什么是负载均衡(Load balancing) 在网站创立初期,我们一般都使用单台机器对台提供集中式服务,但是随着业务量越来越大,无论是性能上还是稳定性上都有了更大的挑战。这时候我们就会想到通过扩容的方式来提供更好的服…

Bumblebee微服务网关之并发限制

对于服务应用来说支持的并发越高越好,但很多时候资源有限,超负载的并发则会给整体应用带来更大的危险性(更何况有些并发来源是恶意的)。作为微服务网关应该具有一定的挡洪作用,这样可以一定程度保障后台逻辑服务的稳定…

[ASP.NET Core 3框架揭秘] 跨平台开发体验: Mac OS

除了微软自家的Windows平台, .NET Core针对Mac OS以及各种Linux Distribution(RHEL、Ubuntu、Debian、Fedora、CentOS和SUSE等)都提供了很好的支持。我们先来体验一下使用Mac来开发.NET Core应用,在这之前我们照例先得在Mac OS上构…

接雨水

题目描述 给定 n 个非负整数表示每个宽度为 1 的柱子的高度图,计算按此排列的柱子,下雨之后能接多少雨水。 上面是由数组 [0,1,0,2,1,0,1,3,2,1,2,1] 表示的高度图,在这种情况下,可以接 6 个单位的雨水(蓝色部分表示…

使用RabbitMQ实现接口补偿

业务背景在我们的日常开发中,经常需要调用第三方接口来进行数据传递,在调用接口的过程中,会因为各种原因导致调用的失败。这时我们希望能有一种机制实现对失败的接口的重复调用,并且能够实现人工干预。实现思路1、当接口调用失败&…

集群环境下,你不得不注意的ASP.NET Core Data Protection 机制

引言最近线上环境遇到一个问题,就是ASP.NET Core Web应用在单个容器使用正常,扩展多个容器无法访问的问题。查看容器日志,发现以下异常:System.Security.Cryptography.CryptographicException: The key {efbb9f35-3a49-4f7f-af19-…

.NET斗鱼直播弹幕客户端(上)

前言现在直播平台由于弹幕的存在,主播与观众可以更轻松地进行互动,非常受年轻群众的欢迎。斗鱼TV就是一款非常流行的直播平台,弹幕更是非常火爆。看到有不少主播接入 弹幕语音播报器、 弹幕点歌等模块,这都需要首先连接斗鱼弹幕。…

程序员后期,架构师发展路线!

作者:zollty,资深程序员和架构师,私底下是个爱折腾的技术极客,架构师社区合伙人!我总结了3个阶段。先说一下各个阶段的感受:1、系统架构阶段:系统架构实际上包括了 业务功能架构 和 技术功能架构。业务上&a…

YUV格式学习

转载自http://blog.csdn.net/searchsun/article/details/2443867 YUV是指亮度参量和色度参量分开表示的像素格式,而这样分开的好处就是不但可以避免相互干扰,还可以降低色度的采样率而不会对图像质量影响太大。YUV是一个比较笼统地说法,针对它…

Flappy Bird游戏python完整源码

通过pygame实现当年风靡一时的flappy bird小游戏。 当前只设定了同样长度的管道,图片和声音文件自行导入。 效果如下: # -*- coding:utf-8 -*- """ 通过pygame实现曾风靡一时的flappybird游戏。 小鸟x坐标不变,画布左移实现…

【 .NET Core 3.0 】框架之九 || 依赖注入 与 IoC

本文有配套视频:https://www.bilibili.com/video/av58096866/?p5前言1、重要:如果你实现了解耦,也就是 api 层只引用了 IService 和 IRepository 的话,那每次修改 service 层,都需要清理解决方案,重新编译…