Go语言定时任务的实现示例

# v1.2
go get github.com/robfig/cron
# v3
go get github.com/robfig/cron/v3@v3.0.0

package main
import (
    "fmt"
    "github.com/robfig/cron/v3"
    "time"
)
func main() {
    // 新建一个定时任务对象,根据cron表达式进行时间调度,cron可以精确到秒,大部分表达式格式也是从秒开始
    // 默认从分开始进行时间调度
    // cronTab := cron.New()
    // 精确到秒
    cronTab := cron.New(cron.WithSeconds())
    // 定义定时器调用的任务函数
    task := func() {
        fmt.Println("hello world", time.Now())
    }
    // 定时任务,cron表达式,每五秒一次
    spec := "*/5 * * * * ?"
    // 添加定时任务
    cronTab.AddFunc(spec, task)
    // 启动定时器
    cronTab.Start()
    // 阻塞主线程停止
    select {}
}

@every <duration>

// 例如
c := cron.New() 
c.AddFunc("@every 1h30m", func() { fmt.Println("Every hour thirty, starting an hour thirty from now") })

package main
import (
    "fmt"
    "github.com/robfig/cron/v3"
    "time"
)
func main() {
    //直接配置时区
    nyc, _ := time.LoadLocation("America/New_York")
    // cron.New(cron.WithLocation(time.UTC))
    c := cron.New(cron.WithLocation(nyc),cron.WithSeconds())
    c.AddFunc("*/5 * * * * ?", func() {
        fmt.Println("Every 5 second at New York")
    })
    // 参数里面配置时区
    c.AddFunc("CRON_TZ=Asia/Tokyo */5 * * * * ?", func() {
        fmt.Println("Every 5 second at Tokyo")
    })
    c.Start()
    select {}
}

package main
import (
    "fmt"
    "github.com/robfig/cron/v3"
    "time"
)
type Task1 struct {
    Name string
}
// 自定义定时任务只需要实现Job接口的Run方法
func (t *Task1) Run() {
    fmt.Println("Task1: ", t.Name)
}
type Task2 struct {
    Name string
}
// 自定义定时任务只需要实现Job接口的Run方法
func (t *Task2) Run() {
    fmt.Println("Task2: ", t.Name)
}
func main() {
    cronTab := cron.New(cron.WithSeconds())
    // 定义定时器调用的任务函数
    // 定时任务
    // cron表达式,每五秒一次
    spec := "*/5 * * * * ?"
    //定义定时器调用的任务函数
    task := func() {
        fmt.Println("hello world", time.Now())
    }
    // 添加多个定时器
    cronTab.AddFunc(spec, task)
    cronTab.AddJob(spec, &Task1{Name: "tom"})
    cronTab.AddJob(spec, &Task2{Name: "merry"})
    // 启动定时器
    cronTab.Start()
    // 关闭,但是不能关闭已经在执行中的任务
    defer cronTab.Stop()
    // 阻塞主线程停止
    select {}
}

// Job is an interface for submitted cron jobs.
type Job interface {
    Run()
}

// 它通过简单的实现Run()方法来实现Job接口
type FuncJob func()
// 这样,任何无参数无返回值的函数,通过强制类型转换为FuncJob,就可以当作Job来使用了,AddFunc方法就是这么做的
func (f FuncJob) Run() { f() }

// Schedule describes a job's duty cycle.
type Schedule interface {
    // Next returns the next activation time, later than the given time.
    // Next is invoked initially, and then each time the job is run.
    // 返回同一Entity中的Job下一次执行的时间
    Next(time.Time) time.Time
}

// SpecSchedule specifies a duty cycle (to the second granularity), based on a
// traditional crontab specification. It is computed initially and stored as bit sets.
type SpecSchedule struct {
    Second, Minute, Hour, Dom, Month, Dow uint64
    // Override location for this schedule.
    Location *time.Location
}

// ConstantDelaySchedule represents a simple recurring duty cycle, e.g. "Every 5 minutes".
// It does not support jobs more frequent than once a second.
type ConstantDelaySchedule struct {
    // 循环的时间间隔
    Delay time.Duration
}

// Next returns the next time this should be run.
// This rounds so that the next activation time will be on the second.
func (schedule ConstantDelaySchedule) Next(t time.Time) time.Time {
    return t.Add(schedule.Delay - time.Duration(t.Nanosecond())*time.Nanosecond)
}

package main
import (
    "fmt"
    "github.com/robfig/cron/v3"
)
func main() {
    constDelaySchedule1 := cron.Every(5e9)
    // 5s
    fmt.Println(constDelaySchedule1.Delay)
    constDelaySchedule2 := cron.Every(5e6)
    // 1s
    fmt.Println(constDelaySchedule2.Delay)
}

// Every returns a crontab Schedule that activates once every duration.
// Delays of less than a second are not supported (will round up to 1 second).
// Any fields less than a Second are truncated.
func Every(duration time.Duration) ConstantDelaySchedule {
    if duration < time.Second {
        duration = time.Second
    }
    return ConstantDelaySchedule{
        Delay: duration - time.Duration(duration.Nanoseconds())%time.Second,
    }
}

// Entry consists of a schedule and the func to execute on that schedule.
type Entry struct {
    // ID is the cron-assigned ID of this entry, which may be used to look up a
    // snapshot or remove it.
    ID EntryID
    // Schedule on which this job should be run.
      // 负责调度当前Entity中的Job执行
    Schedule Schedule
    // Next time the job will run, or the zero time if Cron has not been
    // started or this entry's schedule is unsatisfiable
      // Job下一次执行的时间
    Next time.Time
    // Prev is the last time this job was run, or the zero time if never.
      // 上一次执行时间
    Prev time.Time
    // WrappedJob is the thing to run when the Schedule is activated.
    WrappedJob Job
    // Job is the thing that was submitted to cron.
    // It is kept around so that user code that needs to get at the job later,
    // e.g. via Entries() can do so.
      // 要执行的Job
    Job Job
}

// Cron keeps track of any number of entries, invoking the associated func as
// specified by the schedule. It may be started, stopped, and the entries may
// be inspected while running.
type Cron struct {
    entries   []*Entry
    chain     Chain
    stop      chan struct{} // 控制Cron实例暂停
    add       chan *Entry    // 当Cron已经运行了,增加新的Entity是通过add这个channel实现的
    remove    chan EntryID
    snapshot  chan chan []Entry // 获取当前所有entity的快照
    running   bool // 当已经运行时为true,否则为false
    logger    Logger
    runningMu sync.Mutex
    location  *time.Location
    parser    Parser
    nextID    EntryID
    jobWaiter sync.WaitGroup
}

// Remove an entry from being run in the future.
func (c *Cron) Remove(id EntryID) {
    c.runningMu.Lock()
    defer c.runningMu.Unlock()
    if c.running {
        c.remove <- id
    } else {
        c.removeEntry(id)
    }
}
// Schedule adds a Job to the Cron to be run on the given schedule.
// The job is wrapped with the configured Chain.
func (c *Cron) Schedule(schedule Schedule, cmd Job) EntryID {
    c.runningMu.Lock()
    defer c.runningMu.Unlock()
    c.nextID++
    entry := &Entry{
        ID:         c.nextID,
        Schedule:   schedule,
        WrappedJob: c.chain.Then(cmd),
        Job:        cmd,
    }
    if !c.running {
        c.entries = append(c.entries, entry)
    } else {
        c.add <- entry
    }
    return entry.ID
}

// New returns a new Cron job runner, modified by the given options.
//
// Available Settings
//
//   Time Zone
//     Description: The time zone in which schedules are interpreted
//     Default:     time.Local
//
//   Parser
//     Description: Parser converts cron spec strings into cron.Schedules.
//     Default:     Accepts this spec: https://en.wikipedia.org/wiki/Cron
//
//   Chain
//     Description: Wrap submitted jobs to customize behavior.
//     Default:     A chain that recovers panics and logs them to stderr.
//
// See "cron.With*" to modify the default behavior.
// 实例化时,成员使用的基本是默认值
func New(opts ...Option) *Cron {
    c := &Cron{
        entries:   nil,
        chain:     NewChain(),
        add:       make(chan *Entry),
        stop:      make(chan struct{}),
        snapshot:  make(chan chan []Entry),
        remove:    make(chan EntryID),
        running:   false,
        runningMu: sync.Mutex{},
        logger:    DefaultLogger,
        location:  time.Local,
        parser:    standardParser,
    }
    for _, opt := range opts {
        opt(c)
    }
    return c
}
// Start the cron scheduler in its own goroutine, or no-op if already started.
func (c *Cron) Start() {
    c.runningMu.Lock()
    defer c.runningMu.Unlock()
    if c.running {
        return
    }
    c.running = true
    go c.run()
}

// Run the cron scheduler, or no-op if already running.
func (c *Cron) Run() {
    c.runningMu.Lock()
    if c.running {
        c.runningMu.Unlock()
        return
    }
    c.running = true
    c.runningMu.Unlock()
    c.run()
}
// run the scheduler.. this is private just due to the need to synchronize
// access to the 'running' state variable.
func (c *Cron) run() {
    c.logger.Info("start")
    // Figure out the next activation times for each entry.
    now := c.now()
    for _, entry := range c.entries {
        entry.Next = entry.Schedule.Next(now)
        c.logger.Info("schedule", "now", now, "entry", entry.ID, "next", entry.Next)
    }
    for {
        // Determine the next entry to run.
        sort.Sort(byTime(c.entries))
        var timer *time.Timer
        if len(c.entries) == 0 || c.entries[0].Next.IsZero() {
            // If there are no entries yet, just sleep - it still handles new entries
            // and stop requests.
            timer = time.NewTimer(100000 * time.Hour)
        } else {
            timer = time.NewTimer(c.entries[0].Next.Sub(now))
        }
        for {
            select {
            case now = <-timer.C:
                now = now.In(c.location)
                c.logger.Info("wake", "now", now)
                // Run every entry whose next time was less than now
                for _, e := range c.entries {
                    if e.Next.After(now) || e.Next.IsZero() {
                        break
                    }
                    c.startJob(e.WrappedJob)
                    e.Prev = e.Next
                    e.Next = e.Schedule.Next(now)
                    c.logger.Info("run", "now", now, "entry", e.ID, "next", e.Next)
                }
            case newEntry := <-c.add:
                timer.Stop()
                now = c.now()
                newEntry.Next = newEntry.Schedule.Next(now)
                c.entries = append(c.entries, newEntry)
                c.logger.Info("added", "now", now, "entry", newEntry.ID, "next", newEntry.Next)
            case replyChan := <-c.snapshot:
                replyChan <- c.entrySnapshot()
                continue
            case <-c.stop:
                timer.Stop()
                c.logger.Info("stop")
                return
            case id := <-c.remove:
                timer.Stop()
                now = c.now()
                c.removeEntry(id)
                c.logger.Info("removed", "entry", id)
            }
            break
        }
    }
}
// startJob runs the given job in a new goroutine.
func (c *Cron) startJob(j Job) {
    c.jobWaiter.Add(1)
    go func() {
        defer c.jobWaiter.Done()
        j.Run()
    }()
}

// EntryID标识Cron实例中的entry
type EntryID int
// 将job加入Cron中
// 如上所述,该方法只是简单的通过FuncJob类型强制转换cmd,然后调用AddJob方法
func (c *Cron) AddFunc(spec string, cmd func()) (EntryID, error) 
// 将job加入Cron中
// 通过Parse函数解析cron表达式spec的到调度器实例(Schedule),之后调用c.Schedule方法
func (c *Cron) AddJob(spec string, cmd Job) (EntryID, error)
// 获取当前Cron总所有Entities的快照
func (c *Cron) Entries() []*Entry
// Location获取时区位置
func (c *Cron) Location() *time.Location 
// Entry返回给定项的快照,如果找不到则返回nil
func (c *Cron) Entry(id EntryID) Entry
// 删除将来运行的条目
func (c *Cron) Remove(id EntryID) 
// 通过两个参数实例化一个Entity,然后加入当前Cron中
// 注意: 如果当前Cron未运行,则直接将该entity加入Cron中
// 否则,通过add这个成员channel将entity加入正在运行的Cron中
func (c *Cron) Schedule(schedule Schedule, cmd Job) EntryID
// 新启动一个goroutine运行当前Cron
func (c *Cron) Start()
// 通过给stop成员发送一个struct{}{}来停止当前Cron,同时将running置为false
// 从这里知道,stop只是通知Cron停止,因此往channel发一个值即可,而不关心值是多少
// 所以,成员stop定义为空struct
func (c *Cron) Stop()
// 运行cron调度程序,如果已经在运行,则不运行op
func (c *Cron) Run()

package main
import (
    "fmt"
    "time"
)
func main(){
    go func() {
        for true {
            fmt.Println("Hello World!",time.Now())
            time.Sleep(1 * time.Second)
        }
    }()
    select {
    }
}

package main
import (
    "fmt"
    "time"
)
func main() {
    // 创建一个Timer实例,设置2秒后执行任务
    t := time.NewTimer(2 * time.Second)
    // 记得释放Timer资源
    defer t.Stop()
    // 等待Timer到期
    <-t.C
    // 执行任务
    fmt.Println("Task executed at", time.Now())
}

package main
import (
    "fmt"
    "time"
)
func main() {
    done := make(chan bool)
    ticker := time.NewTimer(1 * time.Second)
    go func() {
        for {
            select {
            case <-done:
                ticker.Stop()
                return
            case <-ticker.C:
                fmt.Println("Hello World!")
            }
        }
    }()
    time.Sleep(10 *time.Second)
    done <- true
}

package main
import (
    "fmt"
    "time"
)
func main() {
    fmt.Println("hello world", time.Now())
    // 创建一个定时器
    // 设置7秒后执行一次
    myT := time.NewTimer(7 * time.Second)
    // 重置定时器为1s后执行
    myT.Reset(1 * time.Second)
    <-myT.C
    fmt.Println("hello world", time.Now())
}

package main
import (
    "fmt"
    "time"
)
func main() {
    // 创建一个Ticker实例,每隔1秒执行一次任务
    ticker := time.NewTicker(1 * time.Second)
    // 记得释放Ticker资源
    defer ticker.Stop()
    // 循环处理任务
    for {
        // 等待Ticker的下一次触发
        <-ticker.C
        // 执行任务
        fmt.Println("Task executed at", time.Now())
    }
}

package main
import (
    "fmt"
    "time"
)
func main() {
    // 创建一个定时器,每隔1秒触发一次
    ticker := time.NewTicker(1 * time.Second)
    // 在函数退出时停止定时器
    defer ticker.Stop()
    // timer实现定时器(延迟执行),这个定时器只会触发一次,所以想要执行定时任务需要放在for循环中
    for {
        select {
        // 定时器触发时执行的任务
        case <-ticker.C:
            fmt.Println("hello world", time.Now())
        }
    }
}

package main
import (
    "fmt"
    "time"
)
func main() {
    ticker := time.NewTicker(1 * time.Second)
    go func() {
        for range ticker.C {
            fmt.Println("Hello World!")
        }
    }()
    time.Sleep(10 * time.Second)
    ticker.Stop()
}

package main
import (
    "fmt"
    "time"
)
func main() {
    done := make(chan bool)
    ticker := time.NewTicker(1 * time.Second)
    go func() {
        for {
            select {
            case <-done:
                ticker.Stop()
                return
            case <-ticker.C:
                fmt.Println("Hello World!")
            }
        }
    }()
    time.Sleep(10 *time.Second)
    done <- true
}

go get -u github.com/go-co-op/gocron

s := gocron.NewScheduler(time.UTC)
s.Every(5).Seconds().Do(func(){ ... }) 
s.Every("5m").Do(func(){ ... })
s.Every(5).Days().Do(fu
s.Every(1).Month(1, 2, 3).Do(func(){ ... })
s.Every(1).Day().At("10:30").Do(func(){ ... })
s.Every(1).Day().At("10:30;08:00").Do(func(){ ... })
s.Every(1).Day().At("10:30").At("08:00").Do(func(){ ... })
s.Every(1).MonthLastDay().Do(func(){ ... })
s.Every(2).MonthLastDay().Do(func(){ ... })
s.Cron("*/1 * * * *").Do(task) 
s.StartAsync()
s.StartBlocking()

package main
import (
    "fmt"
    "time"
    "github.com/go-co-op/gocron"
)
func cron1() {
    fmt.Println("cron1",time.Now())
}
func cron2() {
    fmt.Println("cron2",time.Now())
}
func main() {
    timezone, _ := time.LoadLocation("Asia/Shanghai")
    s := gocron.NewScheduler(timezone)
    // 每秒执行一次
    s.Every(1).Seconds().Do(func() {
        go cron1()
    })
    // 每秒执行一次
    s.Every(1).Second().Do(func() {
        go cron2()
    })
    s.StartBlocking()
}