Go语言并发批量处理数据_Golang数据处理实战-Golang-PHP中文网

应使用信号量限流而非无限制启goroutine，避免OOM；批量写入需拼接多行SQL而非单条执行；优先切片分片而非channel传数据；错误需计数、重试、记录，不可忽略或fatal。

go语言并发批量处理数据_golang数据处理实战

并发控制用 `semaphore` 而不是无限制启 `goroutine`

直接在循环里写 go process(item) 看似简单，但数据量一大就会触发 runtime: out of memory 或系统级资源耗尽。Go 的 goroutine 虽轻量，但每个仍占 2KB+ 栈空间，上万并发瞬间吃光内存。

正确做法是用信号量（如 golang.org/x/sync/semaphore）限流：

sem := semaphore.NewWeighted(int64(maxConcurrency))
var wg sync.WaitGroup
for _, item := range items {
    wg.Add(1)
    go func(i Item) {
        defer wg.Done()
        if err := sem.Acquire(context.Background(), 1); err != nil {
            log.Printf("acquire failed: %v", err)
            return
        }
        defer sem.Release(1)
        process(i)
    }(item)
}
wg.Wait()

maxConcurrency 建议设为 CPU 核心数 × 2～5，具体看 process 是 CPU 密集还是 IO 密集
别漏掉 defer sem.Release(1)，否则后续 goroutine 永远拿不到许可
sem.Acquire 可能阻塞或超时，生产环境建议传带超时的 context

批量写入数据库要避免逐条 `INSERT`

用 go 并发处理数据后直接调 db.Exec("INSERT ...", item.ID, item.Name)，本质是把单条 SQL 放到多个连接里跑——不仅没提升吞吐，还放大了连接池压力和事务开销。

真正高效的批量写法是：先分组，再拼 INSERT INTO ... VALUES (...), (...), (...)，一次提交多行：

const batchSize = 100
for i := 0; i

MySQL / SQLite 支持单语句多值插入；PostgreSQL 需用 UNNEST 或 pgx.Batch
注意 values 参数顺序必须和 placeholders 严格对应
过大的 batchSize（如 > 1000）可能触发 MySQL 的 max_allowed_packet 限制

`channel` 传数据不如切片传得快，别为了“看起来并发”硬套

常见反模式：itemsCh := make(chan Item, 1000)，然后一个 goroutine 往里塞，多个 goroutine 从 channel 读。这引入了额外的同步开销和内存拷贝，实测比直接切片分段慢 15%～30%。

win可用的mysql指令skill

安装、修复、更新或打包一个基于 Python 和 PyMySQL 实现的 Windows `mysql` 命令。适用于用户希望在 PowerShell/CMD 中输入 `mysql` 时重定向到 Python 实现、需要交互式输入 MySQL 连接信息、需要同步到当前用户本地 bin 目录，或需要将该能力封装成可公开复用的 Codex skill。

下载

除非需要动态负载均衡（比如任务耗时差异极大），否则优先用预分片 + 启固定 goroutine：

func splitSlice[T any](s []T, n int) [][]T {
    var chunks [][]T
    for i := 0; i  len(s) {
            end = len(s)
        }
        chunks = append(chunks, s[i:end])
    }
    return chunks
}

chunks := splitSlice(items, len(items)/runtime.NumCPU())
var wg sync.WaitGroup
for _, chunk := range chunks {
    wg.Add(1)
    go func(c []Item) {
        defer wg.Done()
        for _, item := range c {
            process(item)
        }
    }(chunk)
}
wg.Wait()

channel 适合做事件通知、结果聚合、或跨 goroutine 协作，不是通用数据搬运工
切片分片后传参，零拷贝（只传指针和长度），cache 局部性更好
如果下游要合并结果，用 sync.Map 或预先分配好结果切片更可控

错误处理不能只靠 `log.Fatal` 或忽略 `err`

并发场景下，一个 goroutine panic 会终止整个程序；而静默忽略 err 会导致数据丢失却毫无感知。

推荐组合策略：失败计数 + 可恢复错误重试 + 不可恢复错误记录后跳过：

var failed atomic.Int64
var mu sync.RWMutex
var errors []error

for _, item := range items {
    go func(i Item) {
        defer func() {
            if r := recover(); r != nil {
                mu.Lock()
                errors = append(errors, fmt.Errorf("panic on %v: %v", i.ID, r))
                mu.Unlock()
                failed.Add(1)
            }
        }()
        if err := processWithRetry(i, 3); err != nil {
            mu.Lock()
            errors = append(errors, fmt.Errorf("fail on %v: %w", i.ID, err))
            mu.Unlock()
            failed.Add(1)
            return
        }
    }(item)
}
// 等待结束后检查 failed.Load() > 0 再决定是否告警