当前位置: 首页 > news >正文

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写的&#xff0c;原本写日志方案用的是PHP的方案&#xff0c;但是&#xff0c;这个方案导致资源消耗一直降不下来&#xff0c;使用了20个CPU。后面考虑使用通过kafka的方案写日志&#xff0c;商城中把产生的日志丢到kafka中&#xff0c;在以go写的项目…...

Vue.js 使用插槽(Slots)优化组件结构

Vue.js 使用插槽&#xff08;Slots&#xff09;优化组件结构 今天我们聊聊 Vue.js 的一个超实用功能——插槽&#xff08;Slots&#xff09;。插槽是 Vue 组件开发中的神器&#xff0c;用好它&#xff0c;你可以让组件变得更灵活、更可复用&#xff0c;还能写出优雅的代码结构…...

Broker如何进行定时心跳发送和故障感知

1.前言 此文章是在儒猿课程中的学习笔记&#xff0c;感兴趣的想看原来的课程可以去咨询儒猿课堂《从0开始带你成为RocketMQ高手》&#xff0c;我本人觉得这个作者还是不错&#xff0c;都是从场景来进行分析&#xff0c;感觉还是挺适合我这种小白的。这块主要都是我自己的学习笔…...

网络安全设备主要有什么

网络安全设备指的肯定是硬件设备了&#xff0c;国内卖安全硬件的没几家&#xff0c;天融信&#xff0c;启明星辰&#xff0c;绿盟&#xff0c;深信服&#xff0c;就这四家卖的比较齐全吧&#xff0c;上它们官网看一下&#xff0c;就知道市面上主要的网络安全设备有哪些了。分类…...

Android Framework WMS全面概述和知识要点

一、概述 定义与作用 在 Android 系统中&#xff0c;WindowManagerService&#xff08;WMS&#xff09;就像是一个大管家&#xff0c;负责管理整个系统的窗口界面。它是 Android Framework 的核心组件之一&#xff0c;处于 system_server 进程内&#xff0c;在 Framework 层占…...

记一次某红蓝演练经历

在某天接到任务&#xff0c;对xxx进行一次红蓝演练&#xff0c;于是把自己渗透过程给记录下来&#xff0c;漏洞关键地方也会打码&#xff0c;希望各位大佬理解&#xff0c;菜鸡一枚&#xff0c;勿喷/(ㄒoㄒ)/~~ 概述 拿到目标域名第一件事就是信息收集&#xff0c;曾经一位大…...

一个运行在浏览器中的开源Web操作系统Puter本地部署与远程访问

文章目录 前言1.关于Puter2.本地部署Puter3.Puter简单使用4. 安装内网穿透5.配置puter公网地址6. 配置固定公网地址 &#x1f4a1; 推荐 前些天发现了一个巨牛的人工智能学习网站&#xff0c;通俗易懂&#xff0c;风趣幽默&#xff0c;忍不住分享一下给大家。【点击跳转到网站…...

【零基础入门Go语言】struct 和 interface:Go语言是如何实现继承的?

提到面向对象编程中的继承&#xff0c;许多人脑海中可能会浮现出 Java、C 等语言中那一套熟悉的类继承体系。然而&#xff0c;Go 语言作为一门别具一格的编程语言&#xff0c;并没有遵循传统的继承模式。那么&#xff0c;在 Go 语言的世界里&#xff0c;它是怎样实现类似于继承…...

麦田物语学习笔记:实现拖拽物品交换数据和在地图上生成物品

基本流程 1.代码思路 (1)InventoryUI的PlayerSlots与PlayerBag里一一对应,所以想要实现交换数据实际上是,先拿到被拖拽的物体所对的Slot的序号和目标的Slot序号,然后将这两个序号对调一下 (2)物品交换的数据逻辑应该在InventoryManager里去调用,因为InventoryManager里管理了p…...

一些计算机零碎知识随写(25年1月)-1

我原以为世界上有技术的那批人不会那么闲&#xff0c;我错了&#xff0c;被脚本真实了。 今天正隔着画画呢&#xff0c;手机突然弹出几条安全告警通知。 急忙打开服务器&#xff0c;发现问题不简单&#xff0c;直接关服务器重装系统..... 首先&#xff0c;不要认为小网站&…...

Qt学习笔记第81到90讲

第81讲 串口调试助手实现自动发送 为这个名叫“定时发送”的QCheckBox编写槽函数。 想要做出定时发送的效果&#xff0c;必须引入QT框架下的毫秒级定时器QTimer&#xff0c;查阅手册了解详情。 在widget.h内添加新的私有成员变量&#xff1a; QTimer *timer; 在widget类的构造…...

Centos9 + Docker 安装 MySQL8.4.0 + 定时备份数据库到本地

Centos9 Docker 安装 MySQL8.4.0 定时备份数据库到本地 创建目录&#xff0c;创建配置文件启动容器命令定时备份MySQL执行脚本Linux每日定时任务命令文件内参数其他时间参数 AT一次性定时任务 创建目录&#xff0c;创建配置文件 $ mkdir -p /opt/mysql/conf$ vim /opt/mysql/…...

网络原理一>UDP协议详解

UDP和TCP都是应用层中的重要协议&#xff0c;如果做基础架构开发&#xff0c;会用得多一些。 这一篇我们先简单聊一下的UDP TCP格式呈现&#xff1a; 我们知道UDP是一种无连接&#xff0c;面向数据报&#xff0c;全双工&#xff0c;不可靠传输特性的网络协议。 基本格式如图…...

MySQL的小问题

编码问题 不管官方使用什么编码&#xff1a;latin1、gbk、utf8、utfmb4。统一使用utfmb4 MySQL中的utf8并不是utf-8&#xff0c;它省略了一个字节&#xff0c;只是用三个字节存储所有的符号&#xff0c;utfmb4才是utf-8 远程登录问题&#xff1a; MySQL官方默认没有启动远程…...

Mac——Docker desktop安装与使用教程

摘要 本文是一篇关于Mac系统下Docker Desktop安装与使用教程的博文。首先介绍连接WiFi网络&#xff0c;然后详细阐述了如何在Mac上安装Docker&#xff0c;包括下载地址以及不同芯片版本的选择。接着讲解了如何下载基础镜像和指定版本镜像&#xff0c;旨在帮助用户在Mac上高效使…...

FastApi Swagger 序列化问题

问题 错误现象&#xff1a; fastapi的 swagger 界面无法正常打开控制台报错&#xff1a;raise PydanticInvalidForJsonSchema(fCannot generate a JsonSchema for {error_info}) 详细报错&#xff1a; File "d:\Envs\miniconda3\envs\xdagent\lib\site-packages\pydan…...

《机器学习》——sklearn库中CountVectorizer方法(词频矩阵)

CountVectorizer方法介绍 CountVectorizer 是 scikit-learn 库中的一个工具&#xff0c;它主要用于将文本数据转换为词频矩阵&#xff0c;而不是传统意义上的词向量转换&#xff0c;但可以作为词向量转换的一种基础形式。用于将文本数据转换为词频矩阵&#xff0c;它是文本特征…...

UML系列之Rational Rose笔记三:活动图(泳道图)

一、新建活动图&#xff08;泳道图&#xff09; 依旧在用例视图里面&#xff0c;新建一个activity diagram&#xff1b;新建好之后&#xff0c;就可以绘制活动图了&#xff1a; 正常每个活动需要一个开始&#xff0c;点击黑点&#xff0c;然后在图中某个位置安放&#xff0c;接…...

Java面向对象面经总结

目录 面向对象基础 面向对象与面向过程的区别 创建一个对象用什么运算符&#xff0c;对象实体与对象引用的区别 对象相等和引用相等的区别 构造方法的特点&#xff0c;是否可被重写&#xff1f; 面向对象三大特征 封装 继承 多态 接口和抽象类的共同点和区别 深拷贝…...

红队工具使用全解析:揭开网络安全神秘面纱一角

红队工具使用全解析&#xff1a;揭开网络安全神秘面纱一角 B站红队公益课&#xff1a;https://space.bilibili.com/350329294 学习网盘资源链接&#xff1a;https://pan.quark.cn/s/4079487939e8 嘿&#xff0c;各位网络安全爱好者们&#xff01;在风云变幻的网络安全战场上&am…...

K8S认证|CKS题库+答案| 11. AppArmor

目录 11. AppArmor 免费获取并激活 CKA_v1.31_模拟系统 题目 开始操作&#xff1a; 1&#xff09;、切换集群 2&#xff09;、切换节点 3&#xff09;、切换到 apparmor 的目录 4&#xff09;、执行 apparmor 策略模块 5&#xff09;、修改 pod 文件 6&#xff09;、…...

对WWDC 2025 Keynote 内容的预测

借助我们以往对苹果公司发展路径的深入研究经验&#xff0c;以及大语言模型的分析能力&#xff0c;我们系统梳理了多年来苹果 WWDC 主题演讲的规律。在 WWDC 2025 即将揭幕之际&#xff0c;我们让 ChatGPT 对今年的 Keynote 内容进行了一个初步预测&#xff0c;聊作存档。等到明…...

【Zephyr 系列 10】实战项目:打造一个蓝牙传感器终端 + 网关系统(完整架构与全栈实现)

🧠关键词:Zephyr、BLE、终端、网关、广播、连接、传感器、数据采集、低功耗、系统集成 📌目标读者:希望基于 Zephyr 构建 BLE 系统架构、实现终端与网关协作、具备产品交付能力的开发者 📊篇幅字数:约 5200 字 ✨ 项目总览 在物联网实际项目中,**“终端 + 网关”**是…...

【Java_EE】Spring MVC

目录 Spring Web MVC ​编辑注解 RestController RequestMapping RequestParam RequestParam RequestBody PathVariable RequestPart 参数传递 注意事项 ​编辑参数重命名 RequestParam ​编辑​编辑传递集合 RequestParam 传递JSON数据 ​编辑RequestBody ​…...

让AI看见世界:MCP协议与服务器的工作原理

让AI看见世界&#xff1a;MCP协议与服务器的工作原理 MCP&#xff08;Model Context Protocol&#xff09;是一种创新的通信协议&#xff0c;旨在让大型语言模型能够安全、高效地与外部资源进行交互。在AI技术快速发展的今天&#xff0c;MCP正成为连接AI与现实世界的重要桥梁。…...

MacOS下Homebrew国内镜像加速指南(2025最新国内镜像加速)

macos brew国内镜像加速方法 brew install 加速formula.jws.json下载慢加速 &#x1f37a; 最新版brew安装慢到怀疑人生&#xff1f;别怕&#xff0c;教你轻松起飞&#xff01; 最近Homebrew更新至最新版&#xff0c;每次执行 brew 命令时都会自动从官方地址 https://formulae.…...

c++第七天 继承与派生2

这一篇文章主要内容是 派生类构造函数与析构函数 在派生类中重写基类成员 以及多继承 第一部分&#xff1a;派生类构造函数与析构函数 当创建一个派生类对象时&#xff0c;基类成员是如何初始化的&#xff1f; 1.当派生类对象创建的时候&#xff0c;基类成员的初始化顺序 …...

根目录0xa0属性对应的Ntfs!_SCB中的FileObject是什么时候被建立的----NTFS源代码分析--重要

根目录0xa0属性对应的Ntfs!_SCB中的FileObject是什么时候被建立的 第一部分&#xff1a; 0: kd> g Breakpoint 9 hit Ntfs!ReadIndexBuffer: f7173886 55 push ebp 0: kd> kc # 00 Ntfs!ReadIndexBuffer 01 Ntfs!FindFirstIndexEntry 02 Ntfs!NtfsUpda…...

libfmt: 现代C++的格式化工具库介绍与酷炫功能

libfmt: 现代C的格式化工具库介绍与酷炫功能 libfmt 是一个开源的C格式化库&#xff0c;提供了高效、安全的文本格式化功能&#xff0c;是C20中引入的std::format的基础实现。它比传统的printf和iostream更安全、更灵活、性能更好。 基本介绍 主要特点 类型安全&#xff1a…...

Ubuntu系统多网卡多相机IP设置方法

目录 1、硬件情况 2、如何设置网卡和相机IP 2.1 万兆网卡连接交换机&#xff0c;交换机再连相机 2.1.1 网卡设置 2.1.2 相机设置 2.3 万兆网卡直连相机 1、硬件情况 2个网卡n个相机 电脑系统信息&#xff0c;系统版本&#xff1a;Ubuntu22.04.5 LTS&#xff1b;内核版本…...