消息积压了如何处理?

欢迎大家到我的博客阅读这篇文章。消息积压了如何处理? - 胤凯 (oyto.github.io)

在系统中使用消息队列的时候,消息积压这个问题也经常遇到,并且这个问题还不太好解决。

消息积压的直接原因通常是,系统中的某个部分出现了性能问题,来不及处理上游发送的消息,才会导致消息积压。

下面我们来分析下,在使用消息队列的时候,如何优化代码的性能,避免出现消息积压,以及遇到了消息积压问题,该如何进行处理,最大程度地避免消息积压对业务的影响。

优化性能来避免消息积压

对于绝大多数使用消息队列的业务来说,消息队列本身的处理能力要远大于业务能力的处理能力。主流的一些消息队列的单个节点,消息收发的能力可以达到每秒钟处理几万至几十万条消息的水平,还可以通过水平扩展 Broker 的实例数成倍地提升处理能力。

而一般的业务系统需要处理的业务逻辑远比消息队列复杂,单个节点每秒钟可以处理几百到几千次请求,已经可以算是性能非常好的了。所以,对于消息队列的性能优化,我们更应该关注,在消息的收发两端,我们的业务代码怎么和消息队列配合,达到一个最佳的性能。

发送端性能优化

发送端业务代码的处理性能,实际上和消息队列的关系并不打,因为发送端都是先执行自己的业务逻辑,再发送消息。如果系统发送消息的性能上不去,可以优先检查一下是不是发送消息之前的业务逻辑耗时太多导致的。

对于发送消息的业务逻辑,只要注意设置好合适的并发和批量大小,就可以达到很好的发送性能。

我们之前的文章中讲过 Producer 发送消息的过程,Producer 发消息给 Broker,Broker 收到消息后返回确认响应,这是一次完整的交互。假设这一次交互的平均时延是 1 ms,我们把这 1ms 的时间分解开,它包括了下面这些步骤的耗时:

  • 发送端准备数据、序列号消息、构造请求等逻辑的时间,即发送端在发送网络请求之前的耗时

  • 发送消息和返回响应在网络传输中的耗时

  • Broker 处理消息的时延

如果是单线程发送,每次只发送 1 条消息,那么每秒只能发送 1000ms / 1ms * 1 条 /ms = 1000 条 消息,这种情况下并不能发挥出消息队列的全部实力。

这个时候,我们可以选择增加每次发送消息的批量大小或增加并发,都能成倍地增加性能,具体的选择,还是需要根据业务需求来进行选择。

比如说,你的消息发送端是一个微服务,主要接受 RPC 请求处理在线业务。很自然的,微服务在处理每次请求的时候,就在当前线程直接发送消息就可以了,因为所有 RPC 框架都是多线程支持多并发的,自然也就实现了并行发送消息。并且在线业务比较在意的是请求响应时延,选择批量发送必然会影响 RPC 服务的时延。这种情况,比较明智的方式就是通过并发来提升发送性能。

如果你的系统是一个离线分析系统,离线系统在性能上的需求是什么呢?它不关心时延,更注重整个系统的吞吐量。发送端的数据都是来自于数据库,这种情况就更适合批量发送,你可以批量从数据库读取数据,然后批量来发送消息,同样用少量的并发就可以获得非常高的吞吐量。

消费端性能优化

在使用消息队列的时候,大部分性能问题都出现在消费端,如果消费的速度跟不上发送端生产消息的速度,就会造成消息积压。如果这种性能倒挂的问题是暂时的,那问题不大,等消费端的性能恢复之后,超过发送端的性能,那积压的消息是可以被消费完的。

要是消费端的性能一直低于生产端,时间长了整个系统就会出现问题:要么消息队列的存储被填满无法继续提供服务,要么消息丢失,对于整个系统来说都是严重故障。

所以,这种设计系统的时候,一定要保证消费端的消费性能要高于生产端的发送性能,这样才能保证健康的继续运行。

消费端的性能优化还可以通过水平扩展来处理,即增加消费的并发数来提升总体的消费性能。特别需要注意的一点是,在扩容 Consumer 的实例数量的同时,必须同步扩容主题中的分区(也叫队列)数量,确保 Consumer 的实例数和分区数量是相等的。如果 Consumer 的实例数量超过分区数量,这样的扩容实际上是没有效果的。原因我们之前讲过,因为对于消费者来说,在每个分区上实际上只能支持单线程消费。

有很多消费程序,是这样解决消费慢的问题的:

业务处理逻辑比较慢,很难优化,为了避免消息积压,在收到消息的 OnMessage 方法中,不处理任务业务逻辑,而是把这个消息放到内存队列里就返回了。然后启动多个业务线程,这些线程里是真正处理消息的业务逻辑,这些线程从内存队列里取消息处理,这样它就解决了单个 Consumer 不能并行消费的问题。

这个方法看似很完美,但是如果收消息的节点宕机,在内存队列中还没有来得及处理这些消息就会丢失。关于 ”消息丢失“ 问题,可以查看 《如何确保消息不会丢失》这篇文章。

消息积压了该如何处理?

上述两种情况都是在系统设计上的情况,还有一种情况是在系统正常运转时,没有积压或者少量积压后很快就消费掉了,但是某一个时刻,突然开始消息积压并持续上涨。这种情况需要在短时间内找到消息积压的原因,并迅速解决问题,才不至于影响业务。

能导致消息突然积压,最粗粒度的原因只有两种:要么是发送变快了,要么是消费变慢了。

大部分消息队列都内置了监控功能,只要通过监控数据,很容易确定是哪种原因。如果是单位时间发送的消息增加,比如赶上大促销或者抢购,短时间内不太可能优化消费端的代码来提升消费性能,唯一的办法就是通过扩容消费端的实例数来提升总体的消费能力。

如果短时间内没有足够的服务器资源进行扩容,没办法的办法是,将系统降级,通过关闭一些不重要的业务,减少发送方发送的数据量,最低限度让系统还能正常运转,服务一些重要业务。

还有一种不太常见的情况,你通过监控发现,无论是发送消息的速度还是消费消息的速度和原来都没什么变化,这时候你需要检查一下你的消费端,是不是消费失败导致的一条消息反复消费这种情况比较多,这种情况也会拖慢整个系统的消费速度。

如果监控到消费变慢了,你需要检查你的消费实例,分析一下是什么原因导致消费变慢。优先检查一下日志是否有大量的消费错误,如果没有错误的话,可以通过打印堆栈信息,看一下你的消费线程是不是卡在什么地方不动了,比如触发了死锁或者卡在等待某些资源上了。

小结

这篇文章,我们主要讨论了两个问题,一个是如何在设计系统时从消息队列的收发两端优化系统性能,提前预防消息堆积;另一个问题是,当系统发生消息积压后,该如何处理。

优化收发消息性能,预防积压的方法有两种:增加批量或者增加并发,在发送端两种方法都可以使用,在消费端需要注意的是,增加并发需要同步扩容分区数量,否则是不起作用的。

而对于系统发生消息积压的情况,需要先解决积压再分析原因,快速解决积压的方法就是通过水平扩容增加 Consumer 的实例数量,退一步的办法就是将系统降级,关闭一些不重要的业务,保证重要业务的正常运行。

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

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

相关文章

[深度学习]卷积神经网络的概念,入门构建(代码实例)

# 不再任何人,任何组织的身上倾注任何的感情,或许这就是能活得更开心的办法 0.写在前面: 卷积神经网络的部分在之前就已经有所接触,这里重新更全面地总结一下关于深度学习中卷积神经网络的部分.并且在这里对如何构建代码,一些新的思想和网络做出一点点补充,同时会持续更新一些…

异步任务线程池——最优雅的方式创建异步任务

对于刚刚从校园出来的菜鸡选手很容易写出自以为没问题的屎山代码,可是当上线后就会立即暴露出问题,这说到底还是基础不够扎实!只会背八股文,却不理解,面试头头是道,一旦落地就啥也不是。此处,抛…

使用Pandas进行时间重采样,充分挖掘数据价值

大家好,时间序列数据蕴含着很大价值,通过重采样技术可以提升原始数据的表现形式。本文将介绍数据重采样方法和工具,提升数据可视化技巧。 在进行时间数据可视化时,数据重采样是至关重要且非常有用的,它支持控制数据的…

让资产权利归于建设者:Kiosk使过程变得更简单

区块链凭借着其将人的权利地位置于平台之上的能力,可以重塑互联网,而自托管为个人提供了控制和管理其资产和数据的能力。链上交易支持建设者和客户之间的点对点交易。这些特质联合起来,可以将数字世界从基于价值提取的模式转变为基于价值创造…

kk模组的具体应用场合

KK模组是一种高精度、高刚度的直线模组,广泛应用于各种自动化设备和精密仪器中。以下是KK模组的一些具体应用场合: 1、半导体设备:半导体制造过程中需要使用精密的定位和运动控制设备,KK模组作为一种高精度、高刚度的直线模组&…

进程程序替换与exec系统调用

进程程序替换 进程程序替换是指将一个正在运行的进程替换为另一个可执行程序。它的本质是调用了Linux操作系统中的exec系统调用。而exec系统调用是一个家族函数,例如execl、execv、execle、execve等。它们的共同特点是当当前进程执行到该函数时,就会直接…

栈和队列的初始化,插入,删除,销毁。

目录 题外话 顺序表和链表优缺点以及特点 一.栈的特点 二. 栈的操作 2.1初始化 2.2 栈的销毁 2.3 栈的插入 2.3 输出top 2.4 栈的删除 2.5 输出栈 题外话 顺序表和链表优缺点以及特点 特点:顺序表,逻辑地址物理地址。可以任意访问&#xff0c…

CTF-PWN-tips

文章目录 overflowscanfgetreadstrcpystrcat Find string in gdbgdbgdb peda Binary ServiceFind specific function offset in libc手工自动 Find /bin/sh or sh in library手动自动 Leak stack addressFork problem in gdbSecret of a mysterious section - .tlsPredictable …

HDMI之EDID析义篇

DisplayID Type X Video Timing Data Block 实例 F0 2A 10 93 FF 0E 6F 08 8F 10 93 7F 07 37 04 8F 10该数据来源于SHARP AQUOS-TVE23A 4K144Hz电视机的第3个EDID块(基于HF-EEODB)。 定义 解释 VTDB 1: 3840x2160 144.000009 Hz 16:9 333.216 kHz 1343.527000 MHz (RBv3,h…

python中sklearn库在数据预处理中的详细用法,及5个常用的Scikit-learn(通常简称为 sklearn)程序代码示例

文章目录 前言1. 数据清洗:使用 sklearn.preprocessing 中的 StandardScaler 和 MinMaxScaler 进行数据规范化。2. 缺失值处理:使用 sklearn.impute 中的 SimpleImputer 来填充缺失值。3. 数据编码:使用 sklearn.preprocessing 中的 OneHotEn…

Springboot集成JDBC

1&#xff0c;pom.xml配置jar包 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-jdbc</artifactId> </dependency> 2&#xff0c;配置数据源信息 server:port: 8088spring:datasource:dr…

人类智能的精髓超出了统计概率

处理不确定性好坏的程度是衡量各种智能系统高低的一个重要指标。在处理不确定性时&#xff0c;智能系统需要具备推理、学习和决策的能力&#xff0c;通常使用概率和统计等方法来建模和处理不确定性&#xff0c;以便更好地应对现实世界中的复杂问题。统计概率是基于大量观察和数…