Golang中如何使用goroutine实现一个简单的定时任务调度器

time.Ticker

time.Sleep

select

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

// Task 定义了我们调度器中的一个任务
type Task struct {
    Name     string
    Interval time.Duration // 任务执行间隔
    Run      func(ctx context.Context) error // 任务执行的函数，传入context以便取消
}

// Scheduler 是定时任务的管理器
type Scheduler struct {
    tasks      []*Task
    taskCancel map[string]context.CancelFunc // 用于取消单个任务
    mu         sync.Mutex // 保护tasks和taskCancel
    ctx        context.Context // 主调度器的context
    cancel     context.CancelFunc // 取消主调度器
    wg         sync.WaitGroup // 等待所有任务goroutine结束
}

// NewScheduler 创建一个新的调度器
func NewScheduler() *Scheduler {
    ctx, cancel := context.WithCancel(context.Background())
    return &Scheduler{
        tasks:      make([]*Task, 0),
        taskCancel: make(map[string]context.CancelFunc),
        ctx:        ctx,
        cancel:     cancel,
    }
}

// AddTask 向调度器中添加一个任务
func (s *Scheduler) AddTask(task *Task) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.tasks = append(s.tasks, task)
    fmt.Printf("调度器：任务 '%s' 已添加。\n", task.Name)
}

// Start 启动调度器，所有任务将开始执行
func (s *Scheduler) Start() {
    s.mu.Lock()
    defer s.mu.Unlock()

    fmt.Println("调度器：开始启动所有任务...")
    for _, task := range s.tasks {
        taskCtx, taskCancel := context.WithCancel(s.ctx) // 为每个任务创建独立的context
        s.taskCancel[task.Name] = taskCancel // 存储取消函数以便后续停止单个任务

        s.wg.Add(1)
        go s.runTask(taskCtx, task) // 启动goroutine执行任务
    }
    fmt.Println("调度器：所有任务已启动。")
}

// runTask 是每个任务的具体执行逻辑
func (s *Scheduler) runTask(ctx context.Context, task *Task) {
    defer s.wg.Done()
    ticker := time.NewTicker(task.Interval)
    defer ticker.Stop() // 确保ticker被停止

    fmt.Printf("任务 '%s'：开始运行，每 %v 执行一次。\n", task.Name, task.Interval)

    for {
        select {
        case <-ticker.C: // 定时器触发
            fmt.Printf("任务 '%s'：执行中...\n", task.Name)
            err := task.Run(ctx)
            if err != nil {
                fmt.Printf("任务 '%s'：执行失败：%v\n", task.Name, err)
            } else {
                fmt.Printf("任务 '%s'：执行完成。\n", task.Name)
            }
        case <-ctx.Done(): // 收到停止信号
            fmt.Printf("任务 '%s'：收到停止信号，即将退出。\n", task.Name)
            return
        }
    }
}

// Stop 停止调度器，所有正在运行的任务将收到停止信号并退出
func (s *Scheduler) Stop() {
    fmt.Println("调度器：收到停止信号，正在停止所有任务...")
    s.cancel() // 取消主调度器的context，这将向下传播到所有任务的context
    s.wg.Wait() // 等待所有任务goroutine安全退出
    fmt.Println("调度器：所有任务已停止，调度器退出。")
}

// StopTask 停止调度器中的指定任务
func (s *Scheduler) StopTask(taskName string) {
    s.mu.Lock()
    defer s.mu.Unlock()

    if cancel, ok := s.taskCancel[taskName]; ok {
        fmt.Printf("调度器：正在停止任务 '%s'...\n", taskName)
        cancel() // 取消该任务的context
        delete(s.taskCancel, taskName) // 从map中移除
        // 注意：这里我们不等待wg.Done()，因为任务的wg.Done()是在runTask内部完成的
        // 如果需要精确等待单个任务，需要更复杂的WaitGroup管理
    } else {
        fmt.Printf("调度器：任务 '%s' 未找到或已停止。\n", taskName)
    }
}

func main() {
    scheduler := NewScheduler()

    // 添加第一个任务
    scheduler.AddTask(&Task{
        Name:     "清理日志",
        Interval: 2 * time.Second,
        Run: func(ctx context.Context) error {
            // 模拟一个耗时操作
            select {
            case <-time.After(500 * time.Millisecond):
                fmt.Println("      [清理日志] 实际执行：清理了旧日志文件。")
            case <-ctx.Done():
                fmt.Println("      [清理日志] 实际执行：任务被取消，未完成清理。")
                return ctx.Err()
            }
            return nil
        },
    })

    // 添加第二个任务
    scheduler.AddTask(&Task{
        Name:     "数据同步",
        Interval: 3 * time.Second,
        Run: func(ctx context.Context) error {
            fmt.Println("      [数据同步] 实际执行：正在同步数据...")
            // 模拟一个可能失败的任务
            if time.Now().Second()%2 == 0 {
                return fmt.Errorf("模拟错误：数据源连接失败")
            }
            return nil
        },
    })

    scheduler.Start()

    // 让调度器运行一段时间
    time.Sleep(10 * time.Second)

    // 尝试停止一个任务
    scheduler.StopTask("清理日志")
    time.Sleep(3 * time.Second) // 观察停止后的效果

    scheduler.Stop() // 停止所有任务
}

time.After

time.Sleep

time.Sleep

time.After

time.Sleep(interval)

time.Ticker

context.Context

time.Sleep

time.After

context.Context

context.Context

s.ctx

s.cancel

s.Stop()

s.cancel()

context

context

context

context

context

runTask

select

ctx.Done()

context

ctx.Done()

ticker

return

s.wg.Wait()

s.Start()

context

taskCtx, taskCancel := context.WithCancel(s.ctx)

taskCtx

context

taskCancel

taskCancel

map[string]context.CancelFunc

s.StopTask("清理日志")

taskCancel

taskCtx

context

runTask

select { case <-ctx.Done(): ... }

context.Context