Golang实现高效周期性任务调度的轻量级插件库tasks的使用

Tasks是一个简单易用的Go语言进程内调度器，专注于高频、快速执行的周期性任务。它的目标是支持大规模并发任务执行，同时避免调度器引起的抖动。

关键特性

• 并发执行：每个任务在独立的goroutine中执行，确保即使单个任务执行时间较长也不会影响整体调度准确性
• 优化的goroutine调度：使用Go的time.AfterFunc()函数减少休眠goroutine数量，优化CPU调度
• 灵活的任务间隔：使用time.Duration类型指定间隔时间，提供简单接口和灵活控制
• 延迟任务启动：可以指定任务开始执行的时间
• 一次性任务：通过设置RunOnce标志可以创建只执行一次的任务
• 自定义错误处理：可以定义错误处理函数来处理任务返回的错误

使用示例

基本用法

// 启动调度器
scheduler := tasks.New()
defer scheduler.Stop()

// 添加任务
id, err := scheduler.Add(&tasks.Task{
  Interval: 30 * time.Second,
  TaskFunc: func() error {
    // 在这里放入你的逻辑
    fmt.Println("执行周期性任务")
    return nil
  },
})
if err != nil {
  // 处理错误
  log.Fatal(err)
}

延迟调度

// 添加一个每30天执行一次的任务，从30天后开始
id, err := scheduler.Add(&tasks.Task{
  Interval: 30 * (24 * time.Hour),
  StartAfter: time.Now().Add(30 * (24 * time.Hour)),
  TaskFunc: func() error {
    // 在这里放入你的逻辑
    fmt.Println("执行延迟启动的周期性任务")
    return nil
  },
})
if err != nil {
  // 处理错误
  log.Fatal(err)
}

一次性任务

// 添加一个60秒后执行的一次性任务
id, err := scheduler.Add(&tasks.Task{
  Interval: 60 * time.Second,
  RunOnce:  true,
  TaskFunc: func() error {
    // 在这里放入你的逻辑
    fmt.Println("执行一次性任务")
    return nil
  },
})
if err != nil {
  // 处理错误
  log.Fatal(err)
}

自定义错误处理

// 添加带自定义错误处理的任务
id, err := scheduler.Add(&tasks.Task{
  Interval: 30 * time.Second,
  TaskFunc: func() error {
    // 在这里放入你的逻辑
    if rand.Intn(10) < 3 { // 模拟30%的失败率
      return errors.New("随机错误")
    }
    fmt.Println("任务执行成功")
    return nil
  },
  ErrFunc: func(e error) {
    log.Printf("执行任务%s时发生错误 - %s", id, e)
  },
})
if err != nil {
  // 处理错误
  log.Fatal(err)
}

完整示例Demo

下面是一个完整的示例，展示了tasks库的各种用法：

package main

import (
	"errors"
	"fmt"
	"log"
	"math/rand"
	"time"

	"github.com/madflojo/tasks"
)

func main() {
	// 创建调度器
	scheduler := tasks.New()
	defer scheduler.Stop()

	// 1. 基本周期性任务
	_, err := scheduler.Add(&tasks.Task{
		Interval: 5 * time.Second,
		TaskFunc: func() error {
			fmt.Println("基本周期性任务执行 -", time.Now().Format("15:04:05"))
			return nil
		},
	})
	if err != nil {
		log.Fatal(err)
	}

	// 2. 延迟启动的任务
	_, err = scheduler.Add(&tasks.Task{
		Interval:   10 * time.Second,
		StartAfter: time.Now().Add(15 * time.Second),
		TaskFunc: func() error {
			fmt.Println("延迟启动任务执行 -", time.Now().Format("15:04:05"))
			return nil
		},
	})
	if err != nil {
		log.Fatal(err)
	}

	// 3. 一次性任务
	_, err = scheduler.Add(&tasks.Task{
		Interval: 20 * time.Second,
		RunOnce:  true,
		TaskFunc: func() error {
			fmt.Println("一次性任务执行 -", time.Now().Format("15:04:05"))
			return nil
		},
	})
	if err != nil {
		log.Fatal(err)
	}

	// 4. 带错误处理的任务
	_, err = scheduler.Add(&tasks.Task{
		Interval: 8 * time.Second,
		TaskFunc: func() error {
			if rand.Intn(10) < 3 { // 30%概率失败
				return errors.New("模拟任务失败")
			}
			fmt.Println("带错误处理的任务执行成功 -", time.Now().Format("15:04:05"))
			return nil
		},
		ErrFunc: func(e error) {
			log.Printf("任务执行失败: %v", e)
		},
	})
	if err != nil {
		log.Fatal(err)
	}

	// 让程序运行一段时间以便观察任务执行
	time.Sleep(1 * time.Minute)
	fmt.Println("程序结束")
}

这个示例展示了tasks库的主要功能，包括基本周期性任务、延迟启动任务、一次性任务以及带错误处理的任务。你可以根据需要调整间隔时间和任务逻辑。