Go语言本身对并发编程有很好的支持，通过goroutine可以轻松实现多任务并发，但在需要并发执行大量外部命令的场景下，无限制创建goroutine会带来系统资源占用过高的问题，协程池模式可以有效解决这个问题，合理控制并发数量的同时提升资源利用率。
协程池模式的核心设计思路
协程池的核心目标是控制并发执行的协程数量，避免无限制创建goroutine导致的资源耗尽。整体设计需要包含几个核心部分：任务队列、协程管理器、工作协程、结果收集模块。
- 任务队列：用于存放需要执行的外部命令任务，通常是一个带缓冲的channel，避免任务提交时阻塞
- 协程管理器：负责初始化固定数量的工作协程，控制协程的生命周期，在任务完成后回收协程资源
- 工作协程：从任务队列中获取任务，执行对应的外部命令，将执行结果发送到结果队列
- 结果收集模块：接收所有工作协程返回的执行结果，供上层业务处理
核心模块实现
任务与结果定义
首先定义外部命令任务的结构体和执行结果的结构体，明确任务需要包含的命令内容、参数，结果需要包含执行状态、输出内容、错误信息等。
package main
import (
"context"
"fmt"
"os/exec"
"sync"
"time"
)
// 外部命令任务结构体
type CmdTask struct {
ID int // 任务唯一标识
Command string // 要执行的命令，比如"ls"
Args []string // 命令参数，比如[]string{"-l"}
Timeout time.Duration // 命令执行超时时间
}
// 任务执行结果结构体
type TaskResult struct {
TaskID int // 对应任务的ID
Output string // 命令执行的标准输出内容
Err error // 执行错误信息
Cost time.Duration // 执行耗时
}
协程池结构体定义
协程池需要维护任务队列、结果队列、工作协程数量、运行状态等核心属性，同时提供任务提交、池启动、池关闭等方法。
// 协程池结构体
type CmdPool struct {
taskQueue chan CmdTask // 任务队列
resultQueue chan TaskResult // 结果队列
workerNum int // 工作协程数量
wg sync.WaitGroup // 等待所有工作协程退出
ctx context.Context // 上下文，用于控制协程退出
cancel context.CancelFunc // 取消函数
}
// 创建新的协程池实例
func NewCmdPool(workerNum int, taskQueueSize int, resultQueueSize int) *CmdPool {
ctx, cancel := context.WithCancel(context.Background())
return &CmdPool{
taskQueue: make(chan CmdTask, taskQueueSize),
resultQueue: make(chan TaskResult, resultQueueSize),
workerNum: workerNum,
ctx: ctx,
cancel: cancel,
}
}
工作协程实现
工作协程的核心逻辑是从任务队列中循环获取任务，执行外部命令，将结果发送到结果队列，直到收到退出信号。
// 启动工作协程
func (p *CmdPool) startWorker() {
p.wg.Add(1)
go func() {
defer p.wg.Done()
for {
select {
case <-p.ctx.Done():
// 收到退出信号，结束协程
return
case task, ok := <-p.taskQueue:
if !ok {
// 任务队列关闭，结束协程
return
}
// 执行外部命令
startTime := time.Now()
var output []byte
var err error
// 创建带超时的上下文
cmdCtx, cmdCancel := context.WithTimeout(p.ctx, task.Timeout)
defer cmdCancel()
// 执行命令
cmd := exec.CommandContext(cmdCtx, task.Command, task.Args...)
output, err = cmd.CombinedOutput()
cost := time.Since(startTime)
// 构造结果发送到结果队列
result := TaskResult{
TaskID: task.ID,
Output: string(output),
Err: err,
Cost: cost,
}
select {
case p.resultQueue <- result:
case <-p.ctx.Done():
return
}
}
}
}()
}
协程池启动与任务提交
协程池启动时需要初始化指定数量的工作协程，同时提供任务提交方法和结果获取方法，还需要提供池的关闭方法，用于释放资源。
// 启动协程池
func (p *CmdPool) Start() {
for i := 0; i < p.workerNum; i++ {
p.startWorker()
}
}
// 提交任务到协程池
func (p *CmdPool) SubmitTask(task CmdTask) error {
select {
case p.taskQueue <- task:
return nil
case <-p.ctx.Done():
return fmt.Errorf("协程池已关闭，无法提交任务")
}
}
// 获取任务结果
func (p *CmdPool) GetResult() <-chan TaskResult {
return p.resultQueue
}
// 关闭协程池
func (p *CmdPool) Close() {
// 关闭任务队列，停止接收新任务
close(p.taskQueue)
// 发送取消信号，通知所有工作协程退出
p.cancel()
// 等待所有工作协程退出
p.wg.Wait()
// 关闭结果队列
close(p.resultQueue)
}
完整使用示例
下面是一个完整的使用示例，创建包含3个工作协程的协程池，提交5个执行echo命令的任务，然后收集所有任务的执行结果。
func main() {
// 创建协程池，3个工作协程，任务队列容量10，结果队列容量10
pool := NewCmdPool(3, 10, 10)
// 启动协程池
pool.Start()
// 提交5个任务
for i := 0; i < 5; i++ {
taskID := i + 1
task := CmdTask{
ID: taskID,
Command: "echo",
Args: []string{fmt.Sprintf("hello from task %d", taskID)},
Timeout: 5 * time.Second,
}
err := pool.SubmitTask(task)
if err != nil {
fmt.Printf("提交任务%d失败: %vn", taskID, err)
}
}
// 收集结果
go func() {
for result := range pool.GetResult() {
if result.Err != nil {
fmt.Printf("任务%d执行失败，耗时%v，错误: %vn", result.TaskID, result.Cost, result.Err)
} else {
fmt.Printf("任务%d执行成功，耗时%v，输出: %sn", result.TaskID, result.Cost, result.Output)
}
}
}()
// 等待所有任务执行完成，这里简单等待2秒，实际场景可以用更合理的等待方式
time.Sleep(2 * time.Second)
// 关闭协程池
pool.Close()
}
注意事项
在实际使用该协程池模式时，需要注意几个问题：首先是任务队列和结果队列的容量设置，需要根据实际任务数量和处理速度合理调整，避免队列满导致任务提交阻塞或者结果队列满导致工作协程阻塞。其次是外部命令的超时设置，避免某个命令执行时间过长占用协程资源。最后是协程池的关闭逻辑，需要先关闭任务队列，再等待所有工作协程处理完剩余任务，最后关闭结果队列，避免结果丢失或者协程泄漏。