当前位置：首页 > news >正文

用 Go 实现一个轻量级并发任务调度器（支持限速）

news 来源：原创 2025/4/24 21:06:25

前言

在日常开发中，我们经常会遇到这样的场景：

• 有一堆任务要跑（比如：发请求、处理数据、爬虫等）
• 不希望一次性全部跑完，担心打爆服务端或者被封
• 想要设置并发数、限速，还能控制任务重试、失败记录

那么，能不能用 Go 实现一个“轻量级的并发任务调度器”？——答案是：当然可以！

今天我们就来用 Go 从零实现一个可配置的任务调度器，支持：

• 最大并发数控制（worker pool）
• 每秒请求速率限制（rate limit）
• 简单的失败重试机制
• 支持结果收集与错误输出

效果展示

你可以像这样调用我们的调度器：

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("全部任务完成")

核心组件设计

1. 任务（Task）

我们将每个任务抽象为一个结构体：

type Task struct {fn   func() errorretry int
}

2. 调度器（Scheduler）

负责维护任务队列、worker、速率限制器：

type Scheduler struct {tasks       chan *Taskwg          sync.WaitGrouprateLimiter <-chan time.Time
}

实现代码

下面是完整实现（可以直接复制使用）：

type Task struct {fn    func() errorretry int
}func NewTask(fn func() error) *Task {return &Task{fn: fn, retry: 3}
}type Scheduler struct {tasks       chan *Taskwg          sync.WaitGrouprateLimiter <-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)
}