如何用Golang实现高性能的并发池 手写worker pool的实现原理

type Task struct {
    Payload func() error
}

type WorkerPool struct {
    TaskQueue   chan Task
    WorkerQueue chan chan Task
    Workers     []*Worker
    Quit        chan bool
}

type Worker struct {
    ID          int
    WorkerQueue chan chan Task
    TaskQueue   chan Task
    Quit        chan bool
}

func NewWorker(id int, workerQueue chan chan Task) *Worker {
    return &Worker{
        ID:          id,
        WorkerQueue: workerQueue,
        TaskQueue:   make(chan Task),
        Quit:        make(chan bool),
    }
}

func (w *Worker) Start() {
    go func() {
        for {
            // 将自己的任务队列注册到worker池
            w.WorkerQueue <- w.TaskQueue

            select {
            case task := <-w.TaskQueue:
                // 收到任务，执行
                task.Payload()
            case <-w.Quit:
                // 收到退出信号，结束worker
                return
            }
        }
    }()
}

func (w *Worker) Stop() {
    go func() {
        w.Quit <- true
    }()
}

func NewWorkerPool(workerNum int, taskQueueSize int) *WorkerPool {
    taskQueue := make(chan Task, taskQueueSize)
    workerQueue := make(chan chan Task, workerNum)
    workers := make([]*Worker, workerNum)

    pool := &WorkerPool{
        TaskQueue:   taskQueue,
        WorkerQueue: workerQueue,
        Workers:     workers,
        Quit:        make(chan bool),
    }

    for i := 0; i < workerNum; i++ {
        worker := NewWorker(i+1, workerQueue)
        workers[i] = worker
        worker.Start()
    }

    go pool.dispatch()

    return pool
}

func (p *WorkerPool) dispatch() {
    for {
        select {
        case task := <-p.TaskQueue:
            // 从任务队列中取出任务
            workerTaskQueue := <-p.WorkerQueue
            // 将任务发送给worker
            workerTaskQueue <- task
        case <-p.Quit:
            // 收到退出信号，停止所有worker
            for _, worker := range p.Workers {
                worker.Stop()
            }
            return
        }
    }
}

func (p *WorkerPool) Run(task Task) {
    p.TaskQueue <- task
}

func (p *WorkerPool) Stop() {
    go func() {
        p.Quit <- true
    }()
}

// 示例任务
task := Task{
    Payload: func() error {
        // 模拟耗时操作
        time.Sleep(time.Second)
        fmt.Println("Task executed")
        return nil
    },
}

// 提交任务
pool.Run(task)