前言
在日常开发中，我们经常会遇到这样的场景：
- • 有一堆任务要跑（比如：发请求、处理数据、爬虫等）
- • 不希望一次性全部跑完，担心打爆服务端或者被封
- • 想要设置并发数、限速，还能控制任务重试、失败记录
那么，能不能用 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() error retry int }
2. 调度器（Scheduler）
负责维护任务队列、worker、速率限制器：
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) }
实战应用场景
- • 网络爬虫限速抓取
- • 批量发送邮件/SMS/请求，防止接口限流
- • 云服务任务调度、批量自动化操作
- • 异步数据采集和聚合
总结
Go 的并发模型非常适合处理“海量任务 + 控制速率 + 错误重试”的需求。本篇实现的调度器非常轻量，适合作为基础组件集成到你自己的系统中。
如果你有更多需求，比如：
- • 增加失败回调
- • 支持超时控制
- • 任务优先级
- • 后台监控 dashboard
欢迎留言交流，我们可以继续升级这个任务调度器！
关注我，带你用 Go 写出更有趣的小工具！
如果你觉得这篇文章对你有帮助，别忘了点赞、收藏、转发哈～
