Golang协程池实现有问题怎么办？Golang并发模式实践指南

package main

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

type Task func()

type Pool struct {
    size      int
    queue     chan Task
    wg        sync.WaitGroup
    adjusting bool
}

func NewPool(size int) *Pool {
    return &Pool{
        size:  size,
        queue: make(chan Task, 100), // 缓冲大小可调整
    }
}

func (p *Pool) Run() {
    for i := 0; i < p.size; i++ {
        p.wg.Add(1)
        go func() {
            defer p.wg.Done()
            for task := range p.queue {
                task()
            }
        }()
    }
}

func (p *Pool) Submit(task Task) {
    p.queue <- task
}

func (p *Pool) Close() {
    close(p.queue)
    p.wg.Wait()
}

// 简化的动态调整示例
func (p *Pool) AdjustSize(newSize int) {
    if p.adjusting {
        return // 避免并发调整
    }
    p.adjusting = true
    defer func() { p.adjusting = false }()

    if newSize <= 0 {
        fmt.Println("Invalid size")
        return
    }

    oldSize := p.size
    p.size = newSize

    if newSize > oldSize {
        // 增加协程
        for i := 0; i < newSize-oldSize; i++ {
            p.wg.Add(1)
            go func() {
                defer p.wg.Done()
                for task := range p.queue {
                    task()
                }
            }()
        }
    } else if newSize < oldSize {
        // 减少协程 (稍微复杂，需要优雅地停止协程)
        // 这里简化处理，实际应用中需要更完善的停止机制
        // 可以通过发送特殊的task来通知协程退出
        fmt.Println("Shrinking pool size is not fully implemented in this example.")
    }
    fmt.Printf("Pool size adjusted from %d to %d\n", oldSize, newSize)
}

func main() {
    pool := NewPool(runtime.NumCPU())
    pool.Run()

    for i := 0; i < 100; i++ {
        task := func() {
            time.Sleep(100 * time.Millisecond)
            fmt.Println("Task done")
        }
        pool.Submit(task)
    }

    // 模拟一段时间后，调整协程池大小
    time.Sleep(5 * time.Second)
    pool.AdjustSize(runtime.NumCPU() * 2) // 调整为CPU核心数的两倍

    time.Sleep(5 * time.Second)
    pool.Close()
}

package main

import (
    "fmt"
    "time"
)

func worker(id int, jobs <-chan int, results chan<- int) {
    defer fmt.Printf("Worker %d exiting\n", id) // 确保worker退出时打印信息
    for j := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, j)
        time.Sleep(time.Second)
        results <- j * 2
    }
}

func main() {
    numJobs := 5
    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    // 启动3个worker
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    // 发送任务
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs) // 确保jobs通道关闭

    // 收集结果
    for a := 1; a <= numJobs; a++ {
        fmt.Println(<-results)
    }
    close(results) // 确保results通道关闭

    time.Sleep(time.Second) // 等待worker退出
    fmt.Println("All done")
}

package main

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

func safeGo(f func()) {
    go func() {
        defer func() {
            if err := recover(); err != nil {
                // 记录错误信息
                fmt.Printf("panic: %v\n", err)
                // 打印堆栈信息
                buf := make([]byte, 2048)
                runtime.Stack(buf, false)
                fmt.Printf("stack trace:\n%s\n", string(buf))
            }
        }()
        f()
    }()
}

func main() {
    safeGo(func() {
        panic("something went wrong")
    })

    time.Sleep(time.Second) // 等待协程执行
    fmt.Println("Program continues")
}