golang协程池的大小应根据cpu核心数、任务类型、系统资源和压测结果确定。1. cpu核心数：协程池大小不应超过cpu核心数太多，一般为1-2倍；2. 任务类型：cpu密集型任务应接近cpu核心数，i/o密集型任务可适当增加；3. 系统资源：需考虑内存等限制，避免oom；4. 压测：通过测试调整大小，观察吞吐量和响应时间等指标找到最佳平衡点。
协程池，简单来说，就是预先创建好一批协程，需要执行任务时，直接从池子里取一个来用，用完放回去，避免频繁创建和销毁协程的开销。golang标准库本身并没有提供协程池，但我们可以自己实现，或者使用第三方库。
Golang协程池实现方案
实现协程池的核心思路是：维护一个协程队列和一个任务队列。任务来了，就从协程队列里取一个协程去执行，执行完再放回协程队列。
立即学习“go语言免费学习笔记（深入）”；
package main import ( "fmt" "sync" "time" ) type Job struct { ID int Payload int } type WorkerPool struct { JobQueue chan Job WorkerQueue chan chan Job Workers []Worker Quit chan bool Wg sync.WaitGroup } type Worker struct { ID int JobQueue chan Job WorkerQueue chan chan Job Quit chan bool Wg *sync.WaitGroup } func NewWorker(id int, workerQueue chan chan Job, wg *sync.WaitGroup) Worker { return Worker{ ID: id, JobQueue: make(chan Job), WorkerQueue: workerQueue, Quit: make(chan bool), Wg: wg, } } func (w Worker) Start() { w.Wg.Add(1) go func() { defer w.Wg.Done() for { // 将自己的JobChannel 注册到 WorkerPool 的 WorkerQueue 中 w.WorkerQueue <- w.JobQueue select { case job := <-w.JobQueue: // 接收到任务 fmt.Printf("worker%d: 处理 job %d, payload %dn", w.ID, job.ID, job.Payload) time.Sleep(time.Duration(job.Payload) * time.Millisecond) // 模拟耗时操作 fmt.Printf("worker%d: 完成 job %dn", w.ID, job.ID) case <-w.Quit: // 收到停止信号 fmt.Printf("worker%d: 停止n", w.ID) return } } }() } func (w Worker) Stop() { go func() { w.Quit <- true }() } func NewWorkerPool(workerNum int, jobQueueSize int) WorkerPool { jobQueue := make(chan Job, jobQueueSize) workerQueue := make(chan chan Job, workerNum) workers := make([]Worker, workerNum) wp := WorkerPool{ JobQueue: jobQueue, WorkerQueue: workerQueue, Workers: workers, Quit: make(chan bool), Wg: sync.WaitGroup{}, } // 创建 worker for i := 0; i < workerNum; i++ { worker := NewWorker(i+1, wp.WorkerQueue, &wp.Wg) workers[i] = worker } return wp } func (wp WorkerPool) Run() { // 启动所有 worker for i := 0; i < len(wp.Workers); i++ { wp.Workers[i].Start() } go wp.dispatch() } func (wp WorkerPool) dispatch() { for { select { case job := <-wp.JobQueue: // 从 JobQueue 中取出任务 workerJobQueue := <-wp.WorkerQueue // 将任务发送给 Worker workerJobQueue <- job case <-wp.Quit: // 收到停止信号 for i := 0; i < len(wp.Workers); i++ { wp.Workers[i].Stop() } return } } } func (wp WorkerPool) Stop() { close(wp.JobQueue) go func() { wp.Quit <- true }() wp.Wg.Wait() } func main() { workerNum := 5 jobQueueSize := 100 wp := NewWorkerPool(workerNum, jobQueueSize) wp.Run() // 生产 job for i := 0; i < 20; i++ { job := Job{ ID: i + 1, Payload: i*100, // 模拟不同任务的耗时 } wp.JobQueue <- job } // 等待所有 job 完成 time.Sleep(3 * time.Second) wp.Stop() fmt.Println("所有 job 完成") }
Golang 协程池的大小如何确定？
协程池的大小直接影响到程序的并发能力和资源利用率。太小了，并发度不够，浪费资源；太大了，可能导致上下文切换开销过大，甚至OOM。
确定协程池大小需要考虑以下几个因素：
- CPU 核心数： 协程池的大小不应超过 CPU 核心数太多，否则会增加上下文切换的开销。一般来说，可以设置为 CPU 核心数的 1-2 倍。
- 任务类型： 如果任务是 CPU 密集型的，协程池的大小应该接近 CPU 核心数。如果任务是 I/O 密集型的，协程池的大小可以适当增加，因为协程在等待 I/O 时可以切换到其他协程执行。
- 系统资源： 协程池的大小还会受到系统资源的限制，例如内存。如果协程池太大，可能会导致内存不足。
- 压测： 最终，需要通过压测来确定最佳的协程池大小。通过不断调整协程池的大小，并观察程序的性能指标，例如吞吐量、响应时间等，找到一个最佳的平衡点。
Golang 协程池如何处理 panic？
协程中如果发生panic，如果没有recover，会导致程序崩溃。因此，在协程池中处理panic非常重要。
有几种常见的处理方式：
- 在 Worker 中 recover： 这是最常见的方式，在每个 Worker 的执行函数中，使用 recover() 来捕获 panic。这样可以防止 panic 扩散到整个程序，保证协程池的稳定性。
func (w Worker) Start() { w.Wg.Add(1) go func() { defer w.Wg.Done() defer func() { if r := recover(); r != nil { fmt.Printf("worker%d: panic recover: %vn", w.ID, r) // 可以选择将 panic 重新抛出，或者记录日志 } }() for { w.WorkerQueue <- w.JobQueue select { case job := <-w.JobQueue: fmt.Printf("worker%d: 处理 job %d, payload %dn", w.ID, job.ID, job.Payload) // 模拟可能发生 panic 的操作 if job.Payload == 0 { panic("payload is zero") } time.Sleep(time.Duration(job.Payload) * time.Millisecond) fmt.Printf("worker%d: 完成 job %dn", w.ID, job.ID) case <-w.Quit: fmt.Printf("worker%d: 停止n", w.ID) return } } }() }
-
使用第三方库： 一些第三方协程池库，例如 ants，已经内置了 panic 处理机制。使用这些库可以简化 panic 处理的流程。
-
记录日志： 无论使用哪种方式处理 panic，都应该记录详细的日志，包括 panic 的类型、堆栈信息等。这样可以方便后续的排查和修复。
Golang 协程池有哪些常用的第三方库？
虽然可以自己实现协程池，但使用成熟的第三方库可以省去很多麻烦，并获得更好的性能和稳定性。
以下是一些常用的 Golang 协程池第三方库：
- ants： ants 是一个高性能的 Golang 协程池库，它具有以下特点：
- 高性能：基于无锁队列实现，性能优秀。
- 自动调整：可以根据任务负载自动调整协程池的大小。
- panic 处理：内置了 panic 处理机制。
- 资源回收：可以自动回收空闲的协程。
- 使用简单：API 简洁易用。
package main import ( "fmt" "sync" "time" "github.com/panjf2000/ants/v2" ) func main() { defer ants.Release() var wg sync.WaitGroup syncCalculateSum := func(i interface{}) { n := i.(int) fmt.Printf("处理 job %dn", n) time.Sleep(time.Duration(n) * time.Millisecond) fmt.Printf("完成 job %dn", n) wg.Done() } pool, _ := ants.NewPoolWithFunc(10, syncCalculateSum) defer pool.Release() for i := 0; i < 100; i++ { wg.Add(1) _ = pool.Invoke(i) } wg.Wait() fmt.Printf("运行的 goroutine: %dn", ants.Running()) fmt.Printf("完成所有任务.n") }
- tunny： tunny 是另一个流行的 Golang 协程池库，它支持多种任务类型，例如函数、命令等。tunny 的特点是：
- 支持多种任务类型：可以执行函数、命令等。
- 灵活的配置：可以配置协程池的大小、超时时间等。
- 易于扩展：可以自定义 Worker 的行为。
选择哪个第三方库取决于具体的应用场景。如果需要高性能和自动调整，ants 是一个不错的选择。如果需要支持多种任务类型和灵活的配置，tunny 也是一个不错的选择。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
