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

前言

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

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

那么，能不能用 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 &lt;-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)
}