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

【Kafka】登录日志处理的三次阶梯式优化实践：从同步写入到Kafka多分区批处理

news 2025/7/10 12:46:00

登录日志处理的三次阶梯式优化实践：从同步写入到Kafka多分区批处理

本文记录了在GoFrame框架下，针对三端（admin、api、merchant）登录日志处理的三次重大优化历程，展示了如何从同步阻塞到异步高性能处理的技术演进。

优化演进路线

优化阶段	处理方式	吞吐量	延迟	可靠性	扩展性	适用场景
第一阶段	同步写入数据库	低	高	高	低	低流量系统
第二阶段	管道异步批处理	中	中	中	中	中等流量
第三阶段	Kafka多分区消费	高	低	高	高	高并发系统

第一阶段：同步写入数据库

实现方式

在登录逻辑后直接调用数据库写入操作

// 处理用户登录
func Login(ctx context.Context, req *LoginReq) error {// ... 登录验证逻辑 ...// 同步写入登录日志logData := model.LoginLog{UserID:    user.ID,LoginType: req.LoginType,IP:        req.IP,Device:    req.Device,CreatedAt: gtime.Now(),}// 直接插入数据库（同步阻塞）if _, err := dao.LoginLog.Ctx(ctx).Insert(logData); err != nil {g.Log().Errorf(ctx, "写入登录日志失败: %v", err)return err}return nil
}

问题分析

性能瓶颈：数据库写入成为登录流程的瓶颈
高延迟：用户需要等待日志写入完成
低吞吐：无法应对高并发场景

第二阶段：管道异步批处理

架构优化

引入管道(channel)实现生产-消费解耦

// 全局日志管道
var logChan = make(chan *model.LoginLog, 10000)// 初始化日志消费者
func init() {go logConsumer()
}// 日志消费者协程
func logConsumer() {batchSize := 100maxWait := 5 * time.Secondvar batch []*model.LoginLogtimer := time.NewTimer(maxWait)for {select {case logData := <-logChan:batch = append(batch, logData)if len(batch) >= batchSize {flushBatch(batch)batch = niltimer.Reset(maxWait)}case <-timer.C:if len(batch) > 0 {flushBatch(batch)batch = nil}timer.Reset(maxWait)}}
}// 在登录逻辑中发送日志到管道
func Login(ctx context.Context, req *LoginReq) error {// ... 登录验证 ...logData := &model.LoginLog{/*...*/}// 异步发送到管道select {case logChan <- logData:default:// 管道满时的降级处理g.Log().Warning(ctx, "登录日志管道已满，丢弃日志")}return nil
}

优化效果

登录响应时间减少80%
吞吐量提升5倍
通过批处理减少数据库压力

第三阶段：引入Kafka多分区消费

3.1 生产者实现

// Kafka生产者单例
var kafkaProducer *kgo.Client
var producerOnce sync.Once// 获取Kafka生产者（懒加载+单例）
func getKafkaProducer() *kgo.Client {producerOnce.Do(func() {var err errorkafkaProducer, err = kgo.NewClient(kgo.SeedBrokers(global.KafkaConfig.Brokers),kgo.ProducerBatchMaxBytes(5*1024*1024), // 5MB批次kgo.ProducerLinger(100*time.Millisecond), // 等待100ms组批次kgo.RequiredAcks(kgo.AllISRAcks), // 高可靠性)if err != nil {panic(fmt.Sprintf("创建Kafka生产者失败: %v", err))}})return kafkaProducer
}// 发送登录日志到Kafka
func sendLoginLogToKafka(logData *model.LoginLog) error {producer := getKafkaProducer()// 根据用户类型选择分区partition := determinePartition(logData.UserType)jsonData, _ := json.Marshal(logData)record := &kgo.Record{Topic:     "login_log",Value:     jsonData,Partition: partition, // 指定分区}// 异步发送producer.Produce(context.Background(), record, func(r *kgo.Record, err error) {if err != nil {g.Log().Errorf(context.Background(), "发送日志到Kafka失败: %v", err)}})return nil
}// 分区分配策略
func determinePartition(userType string) int32 {switch userType {case "admin": return 0case "api": return 1case "merchant": return 2default: return 0}
}

3.2 消费者实现（多分区批处理）

// Kafka消费者单例
var kafkaConsumer *kgo.Client
var consumerOnce sync.Once// 获取Kafka消费者（懒加载+单例）
func getKafkaConsumer() *kgo.Client {consumerOnce.Do(func() {var err error// 分区配置：三个分区分别对应三种用户类型partitions := map[string]map[int32]kgo.Offset{"login_log": {0: kgo.NewOffset().AtEnd(), // admin分区1: kgo.NewOffset().AtEnd(), // api分区2: kgo.NewOffset().AtEnd(), // merchant分区},}kafkaConsumer, err = kgo.NewClient(kgo.SeedBrokers(global.KafkaConfig.Brokers),kgo.ConsumePartitions(partitions),kgo.FetchMaxBytes(5*1024*1024),     // 最大5MB/次kgo.FetchMaxWait(5*time.Second),    // 最长等待5skgo.FetchMinBytes(1024*1024),       // 至少1MB才返回kgo.MaxConcurrentFetches(3),        // 并发分区数)if err != nil {panic(fmt.Sprintf("创建Kafka消费者失败: %v", err))}})return kafkaConsumer
}// 启动分区消费者
func StartPartitionConsumers() {consumer := getKafkaConsumer()var wg sync.WaitGroup// 为每个分区启动一个消费者协程for partition := 0; partition < 3; partition++ {wg.Add(1)go func(p int32) {defer wg.Done()consumePartition(consumer, p)}(int32(partition))}wg.Wait()
}// 消费指定分区
func consumePartition(client *kgo.Client, partition int32) {const batchSize = 500var batch []*model.LoginLogtimer := time.NewTimer(2 * time.Second)for {select {// 批量处理逻辑case <-timer.C:if len(batch) > 0 {processBatch(batch)batch = nil}timer.Reset(2 * time.Second)default:// 拉取消息fetches := client.PollFetches(context.Background())if fetches.IsClientClosed() {return}// 处理错误if errs := fetches.Errors(); len(errs) > 0 {for _, e := range errs {g.Log().Errorf(context.Background(), "分区%d消费错误: %v", partition, e.Err)}continue}// 处理消息fetches.EachRecord(func(r *kgo.Record) {var logData model.LoginLogif err := json.Unmarshal(r.Value, &logData); err != nil {g.Log().Errorf(context.Background(), "消息解析失败: %v", err)return}batch = append(batch, &logData)// 达到批次大小立即处理if len(batch) >= batchSize {processBatch(batch)batch = niltimer.Reset(2 * time.Second)}})}}
}// 批量处理日志
func processBatch(batch []*model.LoginLog) {// 批量写入数据库if err := batchInsertToDB(batch); err != nil {g.Log().Errorf(context.Background(), "批量写入失败: %v", err)// 失败重试/死信队列处理}metrics.RecordBatchProcessed(len(batch))
}

3.3 关键优化点

分区策略优化

// 分区分配函数
func determinePartition(userType string) int32 {switch userType {case "admin": return 0    // 管理员分区case "api": return 1      // 普通用户分区case "merchant": return 2 // 商家分区default: return 0}
}

消费者并行度

// 每个分区独立消费者
for partition := 0; partition < 3; partition++ {go consumePartition(consumer, int32(partition))
}

批处理参数调优

kgo.FetchMaxBytes(5*1024*1024)  // 5MB/次
kgo.FetchMaxWait(5*time.Second) // 最长等待5s
kgo.FetchMinBytes(1024*1024)    // 至少1MB才返回

客户端复用

// 使用sync.Once确保单例
var consumerOnce sync.Oncefunc getKafkaConsumer() *kgo.Client {consumerOnce.Do(func() {// 初始化逻辑})return kafkaConsumer
}

性能对比数据

指标	同步写入	管道异步	Kafka多分区
平均响应时间	120ms	35ms	15ms
吞吐量	200 TPS	1500 TPS	8000 TPS
CPU占用	高	中	低
数据库压力	极高	高	低
可扩展性	差	一般	优秀

经验总结

分区策略是关键：根据业务特点选择分区策略，我们按用户类型分区实现了并行处理
批处理参数需要调优：
- FetchMinBytes：避免小请求风暴
- FetchMaxWait：平衡延迟和吞吐
- BatchSize：根据数据库写入能力调整

客户端复用很重要：

// 使用sync.Once确保单例
var once sync.Once
var client *kgo.Clientfunc GetKafkaClient() *kgo.Client {once.Do(func() {// 初始化客户端})return client
}

监控不可少：

// 监控批处理指标
func processBatch(batch []*model.LoginLog) {start := time.Now()// ...处理逻辑...duration := time.Since(start)// 记录指标metrics.RecordBatchProcess(len(batch), duration)
}