如何优化Golang的协程创建 控制并发度与协程池实现方案

struct{}

package main

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

func worker(id int) {
    fmt.Printf("Worker %d starting...\n", id)
    time.Sleep(time.Second) // 模拟耗时操作
    fmt.Printf("Worker %d done.\n", id)
}

func main() {
    maxConcurrency := 3 // 最大并发数
    semaphore := make(chan struct{}, maxConcurrency)
    var wg sync.WaitGroup

    for i := 0; i < 10; i++ {
        wg.Add(1)
        semaphore <- struct{}{} // 尝试获取一个令牌，如果通道满则阻塞

        go func(id int) {
            defer wg.Done()
            defer func() { <-semaphore }() // 任务完成后释放令牌

            worker(id)
        }(i)
    }

    wg.Wait()
    fmt.Println("All workers finished.")
    fmt.Printf("Number of goroutines at end: %d\n", runtime.NumGoroutine())
}

semaphore

maxConcurrency

semaphore <- struct{}{}

sync.WaitGroup

WaitGroup

WaitGroup

context.Context

context

package main

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

// Task 定义任务接口，或者直接用函数类型
type Task func()

// Worker 协程，从任务队列中获取任务并执行
func worker(id int, tasks <-chan Task, wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range tasks {
        fmt.Printf("Worker %d processing task...\n", id)
        task() // 执行任务
        time.Sleep(time.Millisecond * 100) // 模拟任务执行
        fmt.Printf("Worker %d finished task.\n", id)
    }
}

// GoroutinePool 协程池结构
type GoroutinePool struct {
    tasks chan Task
    wg    sync.WaitGroup
    size  int
}

// NewGoroutinePool 创建并启动协程池
func NewGoroutinePool(size int) *GoroutinePool {
    pool := &GoroutinePool{
        tasks: make(chan Task), // 任务队列
        size:  size,
    }

    for i := 0; i < size; i++ {
        pool.wg.Add(1)
        go worker(i, pool.tasks, &pool.wg) // 启动Worker协程
    }
    return pool
}

// Submit 提交任务到协程池
func (p *GoroutinePool) Submit(task Task) {
    p.tasks <- task
}

// Close 关闭协程池，等待所有任务完成
func (p *GoroutinePool) Close() {
    close(p.tasks) // 关闭任务队列，Worker协程会退出循环
    p.wg.Wait()    // 等待所有Worker协程完成
    fmt.Println("Goroutine pool closed.")
}

func main() {
    poolSize := 5
    pool := NewGoroutinePool(poolSize)

    for i := 0; i < 20; i++ {
        taskID := i
        pool.Submit(func() {
            fmt.Printf("Executing task %d\n", taskID)
        })
    }

    time.Sleep(time.Second * 3) // 等待一些任务被处理
    pool.Close()
}