Go实现一个轻量级并发任务调度器(支持限速)

scheduler := NewScheduler(5, 10) // 并发 5，速率限制每秒 10 次

for i := 0; i < 100; i++ {
    task := NewTask(func() error {
        // 模拟网络请求或业务逻辑
        fmt.Println("正在处理任务:", i)
        time.Sleep(300 * time.Millisecond)
        return nil
    })
    scheduler.Submit(task)
}

scheduler.Wait()
fmt.Println("全部任务完成")

type Task struct {
    fn   func() error
    retry int
}

type Scheduler struct {
    tasks       chan *Task
    wg          sync.WaitGroup
    rateLimiter <-chan time.Time
}

type Task struct {
    fn    func() error
    retry int
}

func NewTask(fn func() error) *Task {
    return &Task{fn: fn, retry: 3}
}

type Scheduler struct {
    tasks       chan *Task
    wg          sync.WaitGroup
    rateLimiter <-chan time.Time
}

func NewScheduler(concurrency int, ratePerSecond int) *Scheduler {
    s := &Scheduler{
        tasks:       make(chan *Task, 100),
        rateLimiter: time.Tick(time.Second / time.Duration(ratePerSecond)),
    }

    for i := 0; i < concurrency; i++ {
        go s.worker()
    }

    return s
}

func (s *Scheduler) Submit(task *Task) {
    s.wg.Add(1)
    s.tasks <- task
}

func (s *Scheduler) worker() {
    for task := range s.tasks {
        <-s.rateLimiter // 限速

        err := task.fn()
        if err != nil && task.retry > 0 {
            fmt.Println("任务失败，重试中...")
            task.retry--
            s.Submit(task)
        } else if err != nil {
            fmt.Println("任务最终失败:", err)
        }

        s.wg.Done()
    }
}

func (s *Scheduler) Wait() {
    s.wg.Wait()
    close(s.tasks)
}