task

README

Task

A lightweight distributed cron job library for distributed system

用于分布式系统的轻量分布式定时任务库

---

介绍

从 libi/dcron 衍生而来，使用一段时间后发现这个框架存在一些缺陷。

• 不支持本地定时任务
• 节点退出后不能取消注册，在特定场景下会发生任务分配到了已退出的节点上，导致任务无法执行

特性：

在 libi/dcron 基础上，支持：

• 支持本地定时任务
• 支持节点退出取消注册
• 支持懒选举
• 替换robfig/cron为tovenja/cron库，完全兼容且性能更高

Todo：

• 任务执行二次确认
• 完善驱动
• 完善单元测试
• 支持自定义logger

使用

导包

import "cloud/Task"

示例

package main

import (
	"fmt"
	"cloud/Task"
	"taskdriver/redis"
	"time"
)

func main() {
	stop := cronJob()

	defer stop()
}

func cronJob() func() {

	driver := redis.NewDriver(clientRedis())

	cron := Task.NewTask("test-service", driver, Task.WithLazyPick(true))

	// 分布式任务
	_ = cron.AddFunc("job1", Task.JobDistributed, "*/1 * * * *", func() {
		fmt.Println("执行job1: ", time.Now().Format("15:04:05"))
	})

	// 本地任务
	_ = cron.AddFunc("job2", Task.JobLocal, "*/1 * * * *", func() {
		fmt.Println("执行job2: ", time.Now().Format("15:04:05"))
	})

	cron.Start()

	return func() { cron.Stop() }
}

Option说明

• 兼容 tovenja/cron Option

  • WithLocation(loc *time.Location) Option
  • WithSeconds() Option
  • WithParser(p cron.ScheduleParser) Option
  • WithChain(wrappers ...cron.JobWrapper) Option
  • WithLogger(logger cron.Logger) Option

• 自定义节点刷新间隔

  • WithNodeUpdateInterval(dur time.Duration) Option

• 懒选举(任务执行时，拉取节点列表并选举)

  • WithLazyPick(lazy bool) Option

Documentation

Index

• Constants
• type JobWrapper
  • func (job *JobWrapper) Run()
• type Node
  • func (n *Node) Start() error
  • func (n *Node) Stop()
• type Option
  • func WithChain(wrappers ...cron.JobWrapper) Option
  • func WithLazyPick(lazy bool) Option
  • func WithLocation(loc *time.Location) Option
  • func WithLogger(logger cron.Logger) Option
  • func WithNodeUpdateInterval(dur time.Duration) Option
  • func WithParser(p cron.ScheduleParser) Option
  • func WithSeconds() Option
• type Task
  • func NewTask(serviceName string, driver driver.Driver, opts ...Option) *Task
  • func (c *Task) AddFunc(jobName, jobType, spec string, cmd func()) error
  • func (c *Task) AddJob(jobName, jobType, spec string, job cron.Job) error
  • func (c *Task) Remove(jobName string)
  • func (c *Task) Run()
  • func (c *Task) Start()
  • func (c *Task) Stop()

Constants

const (
	JobLocal       = "Local"
	JobDistributed = "Distributed"
)

Types

type JobWrapper

type JobWrapper struct {
	Id   cron.EntryID
	Job  cron.Job
	Func func()

	Task *Task
	Name string
	Type string
}

JobWrapper is a job wrapper

func (*JobWrapper) Run

func (job *JobWrapper) Run()

type Node

type Node struct {
	Task *Task
	// contains filtered or unexported fields
}

func (*Node) Start

func (n *Node) Start() error

func (*Node) Stop

func (n *Node) Stop()

type Option

type Option func(*Task)

func WithChain

func WithChain(wrappers ...cron.JobWrapper) Option

WithChain is wrap cron.Cron with chains

func WithLazyPick

func WithLazyPick(lazy bool) Option

WithLazyPick set lazy pick option

func WithLocation

func WithLocation(loc *time.Location) Option

WithLocation is wrap cron.Cron with location

func WithLogger

func WithLogger(logger cron.Logger) Option

WithLogger is wrap cron.Cron with logger

func WithNodeUpdateInterval

func WithNodeUpdateInterval(dur time.Duration) Option

WithNodeUpdateInterval set node update interval

func WithParser

func WithParser(p cron.ScheduleParser) Option

WithParser is wrap cron.Cron with schedules

func WithSeconds

func WithSeconds() Option

WithSeconds is wrap cron.Cron with seconds

type Task

type Task struct {
	// contains filtered or unexported fields
}

func NewTask

func NewTask(serviceName string, driver driver.Driver, opts ...Option) *Task

func (*Task) AddFunc

func (c *Task) AddFunc(jobName, jobType, spec string, cmd func()) error

AddFunc add a cron func

func (*Task) AddJob

func (c *Task) AddJob(jobName, jobType, spec string, job cron.Job) error

AddJob add a job

func (*Task) Remove

func (c *Task) Remove(jobName string)

Remove Job

func (*Task) Run

func (c *Task) Run()

Run Task

func (*Task) Start

func (c *Task) Start()

Start Task

func (*Task) Stop

func (c *Task) Stop()

Stop Task