Golang并发批量处理技巧与实现

func processBatch(tasks []Task, maxWorkers int) {
    jobs := make(chan Task, len(tasks))
    results := make(chan Result, len(tasks))
<pre class="brush:php;toolbar:false;">// 启动worker
for w := 0; w < maxWorkers; w++ {
    go func() {
        for task := range jobs {
            result := handleTask(task) // 实际处理逻辑
            results <- result
        }
    }()
}

// 发送任务
for _, task := range tasks {
    jobs <- task
}
close(jobs)

// 收集结果
var finalResults []Result
for range tasks {
    finalResults = append(finalResults, <-results)
}
close(results)

}

import "golang.org/x/sync/errgroup"
<p>func processWithErrGroup(ctx context.Context, tasks []Task, limit int) error {
g, ctx := errgroup.WithContext(ctx)
sem := make(chan struct{}, limit) // 控制并发</p><pre class="brush:php;toolbar:false;">for _, task := range tasks {
    task := task
    g.Go(func() error {
        select {
        case sem <- struct{}{}:
            defer func() { <-sem }()
        case <-ctx.Done():
            return ctx.Err()
        }

        return handleTaskWithError(task, ctx)
    })
}

return g.Wait()

}

func batchProcess(tasks []Task, batchSize, concurrency int) {
    for i := 0; i  len(tasks) {
            end = len(tasks)
        }
        batch := tasks[i:end]
<pre class="brush:php;toolbar:false;">    // 处理单个批次
    processBatch(batch, concurrency)
}

}