如何用Golang实现批量并发任务执行

在Golang开发场景中，经常需要处理大量同类型的重复任务，比如批量处理文件、批量调用第三方接口、批量计算数据等。如果采用串行执行的方式，会极大浪费CPU资源，延长整体执行时间，因此通过并发方式批量执行任务是非常必要的。

核心实现思路

Golang实现批量并发任务的核心是利用goroutine创建并发执行单元，结合channel进行任务分发和结果收集，同时需要通过sync包的相关组件控制并发流程，避免协程泄漏或者资源过度占用。

基础并发执行方案

最基础的批量并发实现方式是遍历任务列表，为每个任务启动一个goroutine，然后通过sync.WaitGroup等待所有任务完成。这种方式适合任务数量不多且单个任务资源消耗较小的场景。

package main

import (
	"fmt"
	"sync"
	"time"
)

// 定义任务函数，模拟单个任务的执行
func singleTask(taskID int, wg *sync.WaitGroup) {
	defer wg.Done()
	// 模拟任务执行耗时
	time.Sleep(time.Millisecond * 500)
	fmt.Printf("任务 %d 执行完成n", taskID)
}

func main() {
	// 模拟批量任务列表，共10个任务
	taskCount := 10
	var wg sync.WaitGroup
	// 遍历任务列表，为每个任务启动goroutine
	for i := 0; i < taskCount; i++ {
		wg.Add(1)
		go singleTask(i, &wg)
	}
	// 等待所有任务完成
	wg.Wait()
	fmt.Println("所有批量任务执行完毕")
}

固定协程池方案

如果任务数量非常多，无限制启动goroutine会导致系统资源耗尽，此时可以采用固定协程池的策略，控制同时执行的协程数量。核心思路是创建固定数量的worker协程，通过任务通道分发任务，worker处理完一个任务后继续从通道获取下一个任务。

package main

import (
	"fmt"
	"sync"
	"time"
)

// 任务结构体，可根据实际需求扩展字段
type Task struct {
	ID int
}

// worker函数，处理任务的逻辑
func worker(workerID int, taskChan <-chan Task, wg *sync.WaitGroup) {
	defer wg.Done()
	for task := range taskChan {
		// 模拟任务处理耗时
		time.Sleep(time.Millisecond * 300)
		fmt.Printf("Worker %d 处理了任务 %dn", workerID, task.ID)
	}
}

func main() {
	// 定义任务总数和协程池大小
	taskCount := 20
	poolSize := 5
	taskChan := make(chan Task, taskCount)
	var wg sync.WaitGroup

	// 启动固定数量的worker协程
	for i := 0; i < poolSize; i++ {
		wg.Add(1)
		go worker(i, taskChan, &wg)
	}

	// 向任务通道发送所有任务
	for i := 0; i < taskCount; i++ {
		taskChan <- Task{ID: i}
	}
	close(taskChan)

	// 等待所有worker处理完任务
	wg.Wait()
	fmt.Println("所有任务通过协程池执行完毕")
}

常见批处理策略对比

不同的批处理策略适用不同的场景，下面通过表格对比两种常见方案的优缺点：

注意事项

• 使用sync.WaitGroup时，一定要保证Add的调用次数和Done的调用次数一致，否则会导致程序阻塞或者panic。
• 任务通道使用完毕后要及时关闭，否则worker协程会一直阻塞在读取通道的操作上，造成协程泄漏。
• 如果任务执行过程中可能出现错误，需要额外设计错误收集通道，避免错误被忽略，同时要注意错误通道的关闭时机。
• 并发场景下如果多个任务需要操作共享资源，要通过sync.Mutex或者channel进行同步，避免数据竞争问题。

错误收集示例

在批量并发任务执行时，收集每个任务的错误信息也是常见需求，下面是在固定协程池基础上添加错误收集的实现：

package main

import (
	"fmt"
	"sync"
	"time"
)

type Task struct {
	ID int
}

// 模拟可能出错的任务执行
func executeTask(task Task) error {
	time.Sleep(time.Millisecond * 200)
	// 模拟ID为3的任务执行出错
	if task.ID == 3 {
		return fmt.Errorf("任务 %d 执行失败", task.ID)
	}
	return nil
}

func main() {
	taskCount := 10
	poolSize := 3
	taskChan := make(chan Task, taskCount)
	errChan := make(chan error, taskCount)
	var wg sync.WaitGroup

	// 启动worker，处理任务并收集错误
	for i := 0; i < poolSize; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for task := range taskChan {
				if err := executeTask(task); err != nil {
					errChan <- err
				} else {
					fmt.Printf("任务 %d 执行成功n", task.ID)
				}
			}
		}()
	}

	// 发送任务
	for i := 0; i < taskCount; i++ {
		taskChan <- Task{ID: i}
	}
	close(taskChan)

	// 等待任务处理完成，然后关闭错误通道
	go func() {
		wg.Wait()
		close(errChan)
	}()

	// 收集所有错误
	var errList []error
	for err := range errChan {
		errList = append(errList, err)
	}

	fmt.Printf("执行完成，共 %d 个任务出错n", len(errList))
	for _, err := range errList {
		fmt.Println(err)
	}
}

Golang并发任务批量执行任务批处理goroutine修改时间：2026-07-17 18:36:30