kafka消费堆积问题探索
背景
我们的商城项目用PHP写的,原本写日志方案用的是PHP的方案,但是,这个方案导致资源消耗一直降不下来,使用了20个CPU。后面考虑使用通过kafka的方案写日志,商城中把产生的日志丢到kafka中,在以go写的项目中消费kafka中的日志,并打印到控制台,最后,统一使用阿里sls抓取日志。我们kafka的分区有12个,go程序部署在k8s集群中,开启了弹性扩缩容,最多开启了8个pod进行消费,每秒产生的日志数量高峰在1500条左右,在这种情况下,依然产生了消息的堆积。
消费中执行的逻辑只有对象的映射和日志写控制台,所以,这种情况下产生了消息堆积,令我倍感困惑。
探索之路
第一步,确认一下每一步的执行时间。
func (s KafkaLogService) ReaderCreateLog(ctx context.Context, msg *customerkafka.CustomKafkaMsg) error {now := time.Now().UnixNano()var logEntry LogEntrydata, ok := msg.Data.(string)if !ok {global.GIN_LOG.Error(ctx, "消息数据类型错误", "data", msg.Data)return fmt.Errorf("消息数据类型错误: %v", msg.Data)}// 解析 JSON 数据if err := json.Unmarshal([]byte(data), &logEntry); err != nil {global.GIN_LOG.Error(ctx, "解析 JSON 数据失败:", "error", err, "message", data)return fmt.Errorf("解析消息失败: %w", err)}now2 := time.Now().UnixNano()fmt.Printf("*******************分隔符*******************Unmarshal logEntry 耗时:%d 纳秒\n", now2-now)var logMessage LogMessageif err := json.Unmarshal([]byte(logEntry.Message), &logMessage); err != nil {global.GIN_LOG.Error(ctx, "解析 JSON 数据失败:", "error", err, "message", logEntry.Message)return fmt.Errorf("解析消息失败: %w", err)}now3 := time.Now().UnixNano()fmt.Printf("*******************分隔符*******************Unmarshal LogMessage 耗时:%d 纳秒\n", now3-now2)// 日志等级判断switch logEntry.Level {case "ERROR":global.GIN_LOG.Error(ctx, "", zap.Any("data", logMessage), zap.String("project", logMessage.Project))case "WARN":global.GIN_LOG.Warn(ctx, "", zap.Any("data", logMessage), zap.String("project", logMessage.Project))case "INFO":global.GIN_LOG.Info(ctx, "", zap.Any("data", logMessage), zap.String("project", logMessage.Project))default:global.GIN_LOG.Warn(ctx, "", zap.Any("data", logMessage), zap.String("project", logMessage.Project))}now4 := time.Now().UnixNano()fmt.Printf("*******************分隔符*******************log 耗时:%d 纳秒\n", now4-now3)return nil
}
![]()
![]()
json的Unmarshal的耗时,倒是符合我的认知,在预期之中。
但是,打印日志尽然需要耗时1.5毫秒,这个有点超出我的意料之外。这个时间似乎有点夸张啊。
但是,即便如此,一个消费者每秒也可以消费670条左右的消息,在起了8个实例的情况下,也不应该造成kafka消息的阻塞。
继续我们的探索之路
下面这一段是我对于kakfa消费者的封装。大概的逻辑就是每一个ActionType起一个协程进行消费。在这篇《基于kafka-go写的生产者和消费者》文章中写过这个封装背后的设计逻辑,有兴趣的可以移步过去一探究竟。
// Start 方法启动消费者并开始读取消息,根据actionType调用不同的处理函数
func (c *ConsumerClient) Start(ctx context.Context, handlers map[string]ActionHandler) error {for {select {case <-ctx.Done():return nil // 上下文取消,直接返回default:msg, err := c.reader.ReadMessage(ctx)if err != nil {c.logger.Error(ctx, "Failed to read message from Kafka", "error", err)continue}c.logger.Info(ctx, fmt.Sprintf("Message on topic: %s value: %s partion:%d offset:%d", msg.Topic, string(msg.Value), msg.Partition, msg.Offset))var kafkaMsg CustomKafkaMsgif err := json.Unmarshal(msg.Value, &kafkaMsg); err != nil {c.logger.Error(ctx, "Failed to unmarshal Kafka message", "error", err)continue}channel := make(chan *CustomKafkaMsg)// 使用 sync.Map 来管理 workerworker, loaded := c.workerMap.LoadOrStore(kafkaMsg.ActionType, channel)if !loaded {c.wg.Add(1) // 增加 WaitGroup 计数if ch, ok := worker.(chan *CustomKafkaMsg); ok {go c.startWorker(ctx, kafkaMsg.ActionType, handlers, ch)}}// 发送消息到对应的通道,避免阻塞其他消息消费// 只有在 handlers 中存在对应的 actionType 时才发送消息到对应的通道if _, ok := handlers[kafkaMsg.ActionType]; ok {if ch, ok := worker.(chan *CustomKafkaMsg); ok {ch <- &kafkaMsg}}}}
}
由于我的这个写法,让我产生了一点担忧,虽然,我想的是每个ActionType只起一个协程进行消费,难道,实际情况并不是如我预期一样运行,而是,一条kafka消息就起了一个协程进行消费,如果是这种情况的话,那么,会导致大量的垃圾回收,程序的性能就会下降,那么,消息阻塞的问题也就可以解释了。
为了,验证我的这一想法,基于pprof工具看一下实际情况。

实际验证,排除我的担忧,符合我的预期,不是有一条kafka消息就开一个协程进行消费,而是,一个ActionType就只有一个协程进行消费。
模拟生产环境测试
上述的探索,依然不能够完美解释文章开头提到的现象,起了8个消费者,依然导致消息堆积的现象。为了进一步探究其背后的原因,我模拟生产环境的状态,每秒钟往kafka中丢了1000条消息,再观察,我发现,在这种情况下,有时json.Unmarshal也有比较长的耗时,会出现1.5毫秒的耗时,另外,而写日志需要5毫秒左右,如此,每秒只能消费140条消息,消息堆积的现象也就能够解释了。
![]()
![]()
结论
消息堆积的主要原因是日志打印操作耗时较长,最差时每秒只能消费140条消息。此外,有时JSON解析的时间也较长,这也是一个需要关注的问题。
接下来的目标是找出JSON解析耗时较长和日志打印慢的具体原因,并进行优化。通过解决这些问题,我们有望提高日志处理的效率,从而解决消息堆积的问题。
相关文章:
kafka消费堆积问题探索
背景 我们的商城项目用PHP写的,原本写日志方案用的是PHP的方案,但是,这个方案导致资源消耗一直降不下来,使用了20个CPU。后面考虑使用通过kafka的方案写日志,商城中把产生的日志丢到kafka中,在以go写的项目…...
Vue.js 使用插槽(Slots)优化组件结构
Vue.js 使用插槽(Slots)优化组件结构 今天我们聊聊 Vue.js 的一个超实用功能——插槽(Slots)。插槽是 Vue 组件开发中的神器,用好它,你可以让组件变得更灵活、更可复用,还能写出优雅的代码结构…...
Broker如何进行定时心跳发送和故障感知
1.前言 此文章是在儒猿课程中的学习笔记,感兴趣的想看原来的课程可以去咨询儒猿课堂《从0开始带你成为RocketMQ高手》,我本人觉得这个作者还是不错,都是从场景来进行分析,感觉还是挺适合我这种小白的。这块主要都是我自己的学习笔…...
网络安全设备主要有什么
网络安全设备指的肯定是硬件设备了,国内卖安全硬件的没几家,天融信,启明星辰,绿盟,深信服,就这四家卖的比较齐全吧,上它们官网看一下,就知道市面上主要的网络安全设备有哪些了。分类…...
Android Framework WMS全面概述和知识要点
一、概述 定义与作用 在 Android 系统中,WindowManagerService(WMS)就像是一个大管家,负责管理整个系统的窗口界面。它是 Android Framework 的核心组件之一,处于 system_server 进程内,在 Framework 层占…...
记一次某红蓝演练经历
在某天接到任务,对xxx进行一次红蓝演练,于是把自己渗透过程给记录下来,漏洞关键地方也会打码,希望各位大佬理解,菜鸡一枚,勿喷/(ㄒoㄒ)/~~ 概述 拿到目标域名第一件事就是信息收集,曾经一位大…...
一个运行在浏览器中的开源Web操作系统Puter本地部署与远程访问
文章目录 前言1.关于Puter2.本地部署Puter3.Puter简单使用4. 安装内网穿透5.配置puter公网地址6. 配置固定公网地址 💡 推荐 前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家。【点击跳转到网站…...
【零基础入门Go语言】struct 和 interface:Go语言是如何实现继承的?
提到面向对象编程中的继承,许多人脑海中可能会浮现出 Java、C 等语言中那一套熟悉的类继承体系。然而,Go 语言作为一门别具一格的编程语言,并没有遵循传统的继承模式。那么,在 Go 语言的世界里,它是怎样实现类似于继承…...
麦田物语学习笔记:实现拖拽物品交换数据和在地图上生成物品
基本流程 1.代码思路 (1)InventoryUI的PlayerSlots与PlayerBag里一一对应,所以想要实现交换数据实际上是,先拿到被拖拽的物体所对的Slot的序号和目标的Slot序号,然后将这两个序号对调一下 (2)物品交换的数据逻辑应该在InventoryManager里去调用,因为InventoryManager里管理了p…...
一些计算机零碎知识随写(25年1月)-1
我原以为世界上有技术的那批人不会那么闲,我错了,被脚本真实了。 今天正隔着画画呢,手机突然弹出几条安全告警通知。 急忙打开服务器,发现问题不简单,直接关服务器重装系统..... 首先,不要认为小网站&…...
Qt学习笔记第81到90讲
第81讲 串口调试助手实现自动发送 为这个名叫“定时发送”的QCheckBox编写槽函数。 想要做出定时发送的效果,必须引入QT框架下的毫秒级定时器QTimer,查阅手册了解详情。 在widget.h内添加新的私有成员变量: QTimer *timer; 在widget类的构造…...
Centos9 + Docker 安装 MySQL8.4.0 + 定时备份数据库到本地
Centos9 Docker 安装 MySQL8.4.0 定时备份数据库到本地 创建目录,创建配置文件启动容器命令定时备份MySQL执行脚本Linux每日定时任务命令文件内参数其他时间参数 AT一次性定时任务 创建目录,创建配置文件 $ mkdir -p /opt/mysql/conf$ vim /opt/mysql/…...
网络原理一>UDP协议详解
UDP和TCP都是应用层中的重要协议,如果做基础架构开发,会用得多一些。 这一篇我们先简单聊一下的UDP TCP格式呈现: 我们知道UDP是一种无连接,面向数据报,全双工,不可靠传输特性的网络协议。 基本格式如图…...
MySQL的小问题
编码问题 不管官方使用什么编码:latin1、gbk、utf8、utfmb4。统一使用utfmb4 MySQL中的utf8并不是utf-8,它省略了一个字节,只是用三个字节存储所有的符号,utfmb4才是utf-8 远程登录问题: MySQL官方默认没有启动远程…...
Mac——Docker desktop安装与使用教程
摘要 本文是一篇关于Mac系统下Docker Desktop安装与使用教程的博文。首先介绍连接WiFi网络,然后详细阐述了如何在Mac上安装Docker,包括下载地址以及不同芯片版本的选择。接着讲解了如何下载基础镜像和指定版本镜像,旨在帮助用户在Mac上高效使…...
FastApi Swagger 序列化问题
问题 错误现象: fastapi的 swagger 界面无法正常打开控制台报错:raise PydanticInvalidForJsonSchema(fCannot generate a JsonSchema for {error_info}) 详细报错: File "d:\Envs\miniconda3\envs\xdagent\lib\site-packages\pydan…...
《机器学习》——sklearn库中CountVectorizer方法(词频矩阵)
CountVectorizer方法介绍 CountVectorizer 是 scikit-learn 库中的一个工具,它主要用于将文本数据转换为词频矩阵,而不是传统意义上的词向量转换,但可以作为词向量转换的一种基础形式。用于将文本数据转换为词频矩阵,它是文本特征…...
UML系列之Rational Rose笔记三:活动图(泳道图)
一、新建活动图(泳道图) 依旧在用例视图里面,新建一个activity diagram;新建好之后,就可以绘制活动图了: 正常每个活动需要一个开始,点击黑点,然后在图中某个位置安放,接…...
Java面向对象面经总结
目录 面向对象基础 面向对象与面向过程的区别 创建一个对象用什么运算符,对象实体与对象引用的区别 对象相等和引用相等的区别 构造方法的特点,是否可被重写? 面向对象三大特征 封装 继承 多态 接口和抽象类的共同点和区别 深拷贝…...
红队工具使用全解析:揭开网络安全神秘面纱一角
红队工具使用全解析:揭开网络安全神秘面纱一角 B站红队公益课:https://space.bilibili.com/350329294 学习网盘资源链接:https://pan.quark.cn/s/4079487939e8 嘿,各位网络安全爱好者们!在风云变幻的网络安全战场上&am…...
基于距离变化能量开销动态调整的WSN低功耗拓扑控制开销算法matlab仿真
目录 1.程序功能描述 2.测试软件版本以及运行结果展示 3.核心程序 4.算法仿真参数 5.算法理论概述 6.参考文献 7.完整程序 1.程序功能描述 通过动态调整节点通信的能量开销,平衡网络负载,延长WSN生命周期。具体通过建立基于距离的能量消耗模型&am…...
为什么需要建设工程项目管理?工程项目管理有哪些亮点功能?
在建筑行业,项目管理的重要性不言而喻。随着工程规模的扩大、技术复杂度的提升,传统的管理模式已经难以满足现代工程的需求。过去,许多企业依赖手工记录、口头沟通和分散的信息管理,导致效率低下、成本失控、风险频发。例如&#…...
cf2117E
原题链接:https://codeforces.com/contest/2117/problem/E 题目背景: 给定两个数组a,b,可以执行多次以下操作:选择 i (1 < i < n - 1),并设置 或,也可以在执行上述操作前执行一次删除任意 和 。求…...
鸿蒙DevEco Studio HarmonyOS 5跑酷小游戏实现指南
1. 项目概述 本跑酷小游戏基于鸿蒙HarmonyOS 5开发,使用DevEco Studio作为开发工具,采用Java语言实现,包含角色控制、障碍物生成和分数计算系统。 2. 项目结构 /src/main/java/com/example/runner/├── MainAbilitySlice.java // 主界…...
免费PDF转图片工具
免费PDF转图片工具 一款简单易用的PDF转图片工具,可以将PDF文件快速转换为高质量PNG图片。无需安装复杂的软件,也不需要在线上传文件,保护您的隐私。 工具截图 主要特点 🚀 快速转换:本地转换,无需等待上…...
DingDing机器人群消息推送
文章目录 1 新建机器人2 API文档说明3 代码编写 1 新建机器人 点击群设置 下滑到群管理的机器人,点击进入 添加机器人 选择自定义Webhook服务 点击添加 设置安全设置,详见说明文档 成功后,记录Webhook 2 API文档说明 点击设置说明 查看自…...
云安全与网络安全:核心区别与协同作用解析
在数字化转型的浪潮中,云安全与网络安全作为信息安全的两大支柱,常被混淆但本质不同。本文将从概念、责任分工、技术手段、威胁类型等维度深入解析两者的差异,并探讨它们的协同作用。 一、核心区别 定义与范围 网络安全:聚焦于保…...
算法刷题-回溯
今天给大家分享的还是一道关于dfs回溯的问题,对于这类问题大家还是要多刷和总结,总体难度还是偏大。 对于回溯问题有几个关键点: 1.首先对于这类回溯可以节点可以随机选择的问题,要做mian函数中循环调用dfs(i&#x…...
智警杯备赛--excel模块
数据透视与图表制作 创建步骤 创建 1.在Excel的插入或者数据标签页下找到数据透视表的按钮 2.将数据放进“请选择单元格区域“中,点击确定 这是最终结果,但是由于环境启不了,这里用的是自己的excel,真实的环境中的excel根据实训…...
解决MybatisPlus使用Druid1.2.11连接池查询PG数据库报Merge sql error的一种办法
目录 前言 一、问题重现 1、环境说明 2、重现步骤 3、错误信息 二、关于LATERAL 1、Lateral作用场景 2、在四至场景中使用 三、问题解决之道 1、源码追踪 2、关闭sql合并 3、改写处理SQL 四、总结 前言 在博客:【写在创作纪念日】基于SpringBoot和PostG…...
