Go 并发实战：从零实现一个高并发安全的任务调度器

package scheduler

import (
    "sync"
)

// Task 定义一个无参数的任务函数
type Task func()

// Scheduler 调度器结构
type Scheduler struct {
    taskQueue chan Task   // 任务通道
    workerNum int         // 工作goroutine数量
    wg        sync.WaitGroup // 等待所有工作goroutine结束
    stopChan  chan struct{} // 停止信号通道
}

// NewScheduler 创建一个新的调度器
func NewScheduler(workerNum int) *Scheduler {
    if workerNum <= 0 {
        workerNum = 1
    }
    return &Scheduler{
        taskQueue: make(chan Task, 100), // 缓冲通道，避免阻塞提交者
        workerNum: workerNum,
        stopChan:  make(chan struct{}),
    }
}

// worker 工作goroutine，从通道获取并执行任务
func (s *Scheduler) worker(id int) {
    defer s.wg.Done()
    for {
        select {
        case task, ok := <-s.taskQueue:
            if !ok {
                // 通道已关闭，退出
                return
            }
            // 执行任务，这里可以添加简单的错误处理
            task()
        case <-s.stopChan:
            // 收到停止信号，退出
            return
        }
    }
}

// Start 启动调度器
func (s *Scheduler) Start() {
    for i := 0; i < s.workerNum; i++ {
        s.wg.Add(1)
        go s.worker(i)
    }
}

// Submit 提交一个任务
func (s *Scheduler) Submit(task Task) {
    s.taskQueue <- task
}

// Stop 优雅停止调度器
func (s *Scheduler) Stop() {
    close(s.taskQueue) // 关闭通道，通知worker退出
    <-s.stopChan       // 等待所有worker退出（通过wg）
    s.wg.Wait()
    close(s.stopChan)
}

func main() {
    // 创建一个拥有5个工作goroutine的调度器
    scheduler := NewScheduler(5)
    scheduler.Start()

    // 提交10个任务
    for i := 0; i < 10; i++ {
        scheduler.Submit(func() {
            // 模拟任务执行
            println("Task executed")
        })
    }

    // 等待所有任务完成（这里通过Stop来等待）
    scheduler.Stop()
}