08-第8章-数据队列与重试机制

第8章数据队列与重试机制8.1 队列设计目标8.1.1 为什么需要队列在边缘计算场景中数据可靠性至关重要。队列的作用数据缓冲平滑突发流量削峰填谷故障恢复数据库暂时不可用时不丢失数据离线支持网络中断时缓存数据异步处理解耦数据接收和数据处理8.1.2 队列特性要求特性要求实现方式持久化必须本地文件系统顺序性FIFO文件名按时间排序原子性入队/出队原子互斥锁保护可靠性不丢失数据写入磁盘后返回成功8.2 队列实现原理8.2.1 数据结构// queue/queue.go:32-35typeQueuestruct{queueDirstringmutex sync.Mutex}设计说明queueDir队列数据存储目录mutex互斥锁保证并发安全8.2.2 入队操作// queue/queue.go:50-70func(q*Queue)Enqueue(datainterface{})error{q.mutex.Lock()deferq.mutex.Unlock()// 1. 生成唯一文件名纳秒时间戳filename:fmt.Sprintf(%d.json,time.Now().UnixNano())filepath:filepath.Join(q.queueDir,filename)// 2. 序列化数据为 JSONjsonData,err:json.Marshal(data)iferr!nil{returnfmt.Errorf(failed to marshal data: %v,err)}// 3. 写入文件持久化iferr:os.WriteFile(filepath,jsonData,0644);err!nil{returnfmt.Errorf(failed to write to queue: %v,err)}returnnil}入队流程加锁保证并发安全生成唯一文件名使用纳秒时间戳JSON 序列化数据写入文件系统解锁返回8.2.3 出队操作// queue/queue.go:73-115func(q*Queue)Dequeue()(interface{},error){q.mutex.Lock()deferq.mutex.Unlock()// 1. 读取队列目录files,err:os.ReadDir(q.queueDir)iferr!nil{returnnil,fmt.Errorf(failed to read queue directory: %v,err)}// 2. 找到第一个 JSON 文件FIFOvartargetFile os.DirEntryfor_,file:rangefiles{if!file.IsDir()amp;amp;filepath.Ext(file.Name()).json{targetFilefilebreak}}iftargetFilenil{returnnil,nil// 队列为空}// 3. 读取文件内容filepath:filepath.Join(q.queueDir,targetFile.Name())jsonData,err:os.ReadFile(filepath)iferr!nil{returnnil,fmt.Errorf(failed to read queue file: %v,err)}// 4. 反序列化数据vardatainterface{}iferr:json.Unmarshal(jsonData,amp;data);err!nil{returnnil,fmt.Errorf(failed to unmarshal data: %v,err)}// 5. 删除文件iferr:os.Remove(filepath);err!nil{returnnil,fmt.Errorf(failed to remove queue file: %v,err)}returndata,nil}出队流程加锁保证并发安全读取目录找到最早的文件读取文件内容JSON 反序列化删除文件解锁返回数据8.2.4 获取队列大小// queue/queue.go:118-137func(q*Queue)Size()(int,error){q.mutex.Lock()deferq.mutex.Unlock()files,err:os.ReadDir(q.queueDir)iferr!nil{return0,fmt.Errorf(failed to read queue directory: %v,err)}count:0for_,file:rangefiles{if!file.IsDir()amp;amp;filepath.Ext(file.Name()).json{count}}returncount,nil}8.3 队列处理机制8.3.1 后台处理 Goroutine// queue/queue.go:144-185func(q*Queue)ProcessQueue(processFuncfunc(interface{})error){gofunc(){for{// 1. 检查队列大小size,err:q.Size()iferr!nil{log.Printf(Failed to get queue size: %v,err)time.Sleep(5*time.Second)continue}// 2. 队列为空则等待ifsize0{time.Sleep(5*time.Second)continue}// 3. 出队数据data,err:q.Dequeue()iferr!nil{log.Printf(Failed to dequeue data: %v,err)time.Sleep(5*time.Second)continue}ifdatanil{time.Sleep(5*time.Second)continue}// 4. 调用处理函数iferr:processFunc(data);err!nil{log.Printf(Failed to process queue data: %v,err)// 5. 处理失败重新入队iferr:q.Enqueue(data);err!nil{log.Printf(Failed to re-enqueue data: %v,err)}time.Sleep(5*time.Second)}}}()}8.3.2 主程序中的队列处理器// main.go:105-112dataQueue.ProcessQueue(func(datainterface{})error{records,ok:data.([]*map[string]any)if!ok{returnfmt.Errorf(invalid data type in queue)}// 使用重试机制插入数据returndatabase.BatchInsertWithRetry(database.Table,records,3,2*time.Second)})8.4 重试机制8.4.1 批量插入重试// database/database.go:208-222funcBatchInsertWithRetry(tbl*engine.Table,records[]*map[string]any,maxRetriesint,retryInterval time.Duration)error{fori:0;ilt;maxRetries;i{_,err:tbl.BatchInsertNoInc(records)iferrnil{returnnil}log.Printf(Failed to batch insert data (attempt %d/%d): %v,i1,maxRetries,err)ifilt;maxRetries-1{time.Sleep(retryInterval)}}returnfmt.Errorf(failed to batch insert data after %d attempts,maxRetries)}8.4.2 重试策略策略说明适用场景固定间隔每次重试等待相同时间临时性故障指数退避等待时间指数增长避免雪崩最大重试次数限制重试次数避免无限重试降级处理重试失败后降级保证可用性8.4.3 MQTT 中的重试// mqtt/client.go 中的错误处理_,err:database.Table.BatchInsertNoInc(messages)iferr!nil{log.Printf(Failed to batch insert: %v,err)// 写入队列后续重试iferr:c.dataQueue.Enqueue(messages);err!nil{log.Printf(Failed to enqueue: %v,err)}return}8.5 队列优化8.5.1 性能优化建议批量处理一次处理多条记录减少文件 I/O目录优化使用多个子目录分散文件内存缓存热点数据在内存中缓存异步写入使用 sync.Mutex 保证安全8.5.2 容量规划// 队列大小监控size,err:dataQueue.Size()iferr!nil{log.Printf(Queue size error: %v,err)}elseifsizegt;10000{log.Printf(Warning: Queue size is large: %d,size)// 触发告警monitorInstance.RecordWarning(queue_backlog,fmt.Sprintf(Queue size: %d,size))}8.6 实战练习练习 8.1队列压力测试编写一个压力测试测试队列的吞吐量和延迟。练习 8.2指数退避重试实现指数退避的重试策略。练习 8.3死信队列实现死信队列处理多次重试失败的数据。8.7 本章小结本章深入解析了数据队列和重试机制队列的设计目标和特性要求队列的实现原理入队、出队、大小查询后台队列处理机制重试策略的实现队列优化建议队列和重试机制是保证数据可靠性的关键组件。本书版本1.0.0最后更新2026-03-08sfsEdgeStore- 让边缘数据存储更简单技术栈- Go语言、sfsDb与EdgeX Foundry。纯golang工业物联网边缘计算技术栈项目地址GitHub