go-accumulator 使用指南

go-accumulator 是一个用于事件聚合与批量处理的 Golang 库，它可以帮助开发者高效地收集、聚合和批量处理事件数据。下面我将详细介绍它的使用方法和示例代码。

核心特性

1. 事件聚合：将多个小事件聚合成批量数据
2. 定时触发：支持按时间间隔自动触发处理
3. 大小触发：当累积事件达到指定数量时触发
4. 并发安全：支持多协程并发操作
5. 自定义处理：灵活定义批量数据处理逻辑

安装

go get github.com/go-accumulator/accumulator

基本使用示例

package main

import (
	"fmt"
	"time"
	
	"github.com/go-accumulator/accumulator"
)

func main() {
	// 创建一个新的累加器
	acc := accumulator.NewAccumulator(
		accumulator.WithBatchSize(10),     // 每10条触发一次
		accumulator.WithInterval(5*time.Second), // 或5秒触发一次
		accumulator.WithHandler(handleBatch), // 设置批处理函数
	)
	
	// 启动累加器
	acc.Start()
	
	// 模拟产生事件
	for i := 0; i < 25; i++ {
		acc.Add(i) // 添加事件
		time.Sleep(500 * time.Millisecond)
	}
	
	// 关闭累加器
	acc.Stop()
}

// 批处理函数
func handleBatch(items []interface{}) {
	fmt.Printf("处理批量数据(%d条): %v\n", len(items), items)
}

高级配置选项

acc := accumulator.NewAccumulator(
	accumulator.WithBatchSize(100),          // 每100条触发
	accumulator.WithInterval(30*time.Second), // 或30秒触发
	accumulator.WithHandler(handleBatch),     // 处理函数
	accumulator.WithMaxWait(5*time.Second),  // 最大等待时间
	accumulator.WithBufferSize(1000),        // 缓冲区大小
	accumulator.WithSharding(4),             // 分片数，提高并发性能
)

实际应用场景示例

1. 日志批量上传

func main() {
	// 创建日志累加器
	logAcc := accumulator.NewAccumulator(
		accumulator.WithBatchSize(50),
		accumulator.WithInterval(10*time.Second),
		accumulator.WithHandler(uploadLogs),
	)
	
	// 在HTTP处理中使用
	http.HandleFunc("/log", func(w http.ResponseWriter, r *http.Request) {
		log := r.URL.Query().Get("message")
		if log != "" {
			logAcc.Add(log)
		}
		w.WriteHeader(http.StatusOK)
	})
	
	logAcc.Start()
	http.ListenAndServe(":8080", nil)
}

func uploadLogs(logs []interface{}) {
	// 将日志批量上传到日志服务器
	fmt.Printf("上传%d条日志\n", len(logs))
}

2. 数据库批量插入

func main() {
	dbAcc := accumulator.NewAccumulator(
		accumulator.WithBatchSize(200),
		accumulator.WithInterval(15*time.Second),
		accumulator.WithHandler(batchInsert),
	)
	
	// 模拟数据产生
	for i := 0; i < 1000; i++ {
		record := map[string]interface{}{
			"id":   i,
			"name": fmt.Sprintf("user%d", i),
		}
		dbAcc.Add(record)
	}
	
	dbAcc.Start()
	time.Sleep(1 * time.Minute)
	dbAcc.Stop()
}

func batchInsert(records []interface{}) {
	// 这里实现批量插入数据库逻辑
	fmt.Printf("批量插入%d条记录\n", len(records))
}

性能优化建议

1. 合理设置批大小：根据后端处理能力调整
2. 使用分片：高并发场景下使用WithSharding
3. 监控指标：添加处理耗时和批大小的监控
4. 错误处理：在批处理函数中添加重试机制

注意事项

1. 处理函数应该快速完成，避免阻塞
2. 程序退出前调用Stop()确保所有数据被处理
3. 处理函数中的panic会被捕获并打印，但不会中断程序

go-accumulator 是一个简单而强大的工具，特别适合需要将分散操作聚合成批量处理的场景，能显著提高系统吞吐量和降低I/O压力。