Documentation
¶
Overview ¶
Package scheduler provides a distributed task scheduler with distributed locking and task dispatch.
Unlike a simple cron library (which only parses expressions and fires callbacks in a single process), this package ensures that scheduled jobs run on exactly one node in a cluster — even when multiple instances of the same application are running simultaneously.
Key features ¶
- Cron expression scheduling (via common/cron) and interval scheduling.
- Distributed lock integration (via the lock package): only the node that acquires the lock for a job executes it.
- Pluggable LockFactory: use Redis, etcd, Zookeeper, or the in-memory lock for single-node / testing.
- Context-aware job execution with timeout.
- Job metadata: name, description, tags, singleton mode.
- Graceful shutdown: stop accepting new ticks, wait for running jobs.
- Job status tracking: last run, next run, error count.
- Optional job event listener for monitoring/metrics.
Architecture ¶
The Scheduler runs a goroutine per job. Each goroutine calculates the next fire time from the cron expression (or interval), sleeps until that time, then attempts to acquire a distributed lock for the job. If the lock is acquired, the job function runs. If not (another node got it), the job is skipped and the goroutine waits for the next fire time.
This design is deliberately simple and avoids external dependencies on message queues or coordination services beyond the lock backend.
Quick start (single node) ¶
// In-memory lock — single node only.
lockMgr := memory.NewManager()
s := scheduler.New(scheduler.Config{
LockFactory: scheduler.LockFactoryFunc(func(jobName string) (lock.Locker, error) {
return lockMgr.NewMutex("scheduler:"+jobName, lock.WithTTL(30*time.Second))
}),
})
s.Start()
s.Add("cleanup", "*/5 * * * *", func(ctx context.Context) error {
return cleanupDatabase(ctx)
})
// Graceful shutdown.
s.Stop()
Quick start (distributed, Redis) ¶
rdb := redis.NewClient(...)
s := scheduler.New(scheduler.Config{
LockFactory: scheduler.LockFactoryFunc(func(jobName string) (lock.Locker, error) {
return redis.NewMutex(rdb, "scheduler:"+jobName,
lock.WithTTL(60*time.Second),
lock.WithRetryDelay(200*time.Millisecond))
}),
LockTTL: 60 * time.Second,
})
s.Start()
s.Add("report", "0 * * * *", generateReport)
defer s.Stop()
Index ¶
- Variables
- type Config
- type EventListener
- type EventListenerFunc
- type Job
- type JobEvent
- type JobEventType
- type JobFunc
- type JobStatus
- type LockFactory
- type LockFactoryFunc
- type NoLockFactory
- type Options
- type Scheduler
- func (s *Scheduler) Add(name, schedule string, fn JobFunc) error
- func (s *Scheduler) AddWithOptions(name, schedule string, fn JobFunc, opts Options) error
- func (s *Scheduler) Get(name string) (*Job, error)
- func (s *Scheduler) JobCount() int
- func (s *Scheduler) Jobs() []*Job
- func (s *Scheduler) Remove(name string) error
- func (s *Scheduler) Start()
- func (s *Scheduler) Stop()
Constants ¶
This section is empty.
Variables ¶
var ( // ErrSchedulerStopped is returned when operating on a stopped scheduler. ErrSchedulerStopped = errors.New("scheduler: stopped") // ErrJobNotFound is returned when a job with the given name does not exist. ErrJobNotFound = errors.New("scheduler: job not found") // ErrJobExists is returned when adding a job with a name that already exists. ErrJobExists = errors.New("scheduler: job already exists") // ErrInvalidSchedule is returned when the schedule expression is invalid. ErrInvalidSchedule = errors.New("scheduler: invalid schedule") // ErrNoLockFactory is returned when no LockFactory is configured. ErrNoLockFactory = errors.New("scheduler: no lock factory configured") )
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// LockFactory creates distributed locks for jobs. Required for distributed
// mode. Use NoLockFactory{} for single-node mode (no dispatch).
LockFactory LockFactory
// LockTTL is the TTL for the distributed lock. Default: 30s.
// The lock is refreshed periodically while the job runs (if the lock
// backend supports Refresh).
LockTTL time.Duration
// LockRefreshInterval is how often the lock is refreshed while a job
// runs. Default: LockTTL / 3.
LockRefreshInterval time.Duration
// Timezone is the timezone for cron expression evaluation.
// Default: time.Local.
Timezone *time.Location
// EventListener receives job events. Optional.
EventListener EventListener
}
Config configures the scheduler.
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns a Config with sensible defaults.
type EventListener ¶
type EventListener interface {
OnJobEvent(event JobEvent)
}
EventListener receives job events. Use this for monitoring, metrics, or logging.
type EventListenerFunc ¶
type EventListenerFunc func(JobEvent)
EventListenerFunc is a function adapter for EventListener.
func (EventListenerFunc) OnJobEvent ¶
func (f EventListenerFunc) OnJobEvent(e JobEvent)
OnJobEvent implements EventListener.
type Job ¶
type Job struct {
Name string // unique job identifier
Description string // human-readable description
Schedule string // cron expression or "@every <duration>"
Func JobFunc // the function to execute
Timeout time.Duration // per-execution timeout (0 = no timeout)
Singleton bool // if true, skip if previous run is still active
Tags []string // optional tags for grouping/filtering
// contains filtered or unexported fields
}
Job represents a scheduled task.
type JobEvent ¶
type JobEvent struct {
JobName string
Type JobEventType
Time time.Time
Error error
Duration time.Duration
}
JobEvent represents an event in a job's lifecycle.
type JobEventType ¶
type JobEventType int
JobEventType describes the type of job event.
const ( // EventJobStarted is emitted when a job starts executing. EventJobStarted JobEventType = iota // EventJobSucceeded is emitted when a job completes successfully. EventJobSucceeded // EventJobFailed is emitted when a job returns an error. EventJobFailed // EventJobSkipped is emitted when a job is skipped (lock not acquired // or singleton mode blocked it). EventJobSkipped // EventJobAdded is emitted when a job is added to the scheduler. EventJobAdded // EventJobRemoved is emitted when a job is removed from the scheduler. EventJobRemoved )
type JobFunc ¶
JobFunc is the function executed when a job fires. It receives a context that is cancelled when the job timeout expires or when the scheduler is shutting down.
type JobStatus ¶
type JobStatus struct {
LastRun time.Time
LastEnd time.Time
NextRun time.Time
LastError error
LastErrorAt time.Time
RunCount int64
ErrorCount int64
Running bool
}
JobStatus holds the runtime status of a job.
type LockFactory ¶
type LockFactory interface {
// NewLock returns a Locker for the given job name. The returned Locker
// will be used with TryLock (non-blocking) by the scheduler.
NewLock(jobName string) (lock.Locker, error)
}
LockFactory creates a distributed lock for a given job name. Each job gets its own lock key so that different jobs can run in parallel on different nodes, while the same job runs on only one node at a time.
Implementations can wrap the lock/redis, lock/etcd, lock/memory, etc. backends.
type LockFactoryFunc ¶
LockFactoryFunc is a function adapter for LockFactory.
type NoLockFactory ¶
type NoLockFactory struct{}
NoLockFactory is a LockFactory that returns nil locks. Use this for single-node scheduling where distributed locking is not needed. In this mode, jobs always run on every node (no dispatch).
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler manages a set of scheduled jobs with distributed locking.
func (*Scheduler) Add ¶
Add registers a new job. If the scheduler is already started, the job begins running immediately.
func (*Scheduler) AddWithOptions ¶
AddWithOptions registers a new job with additional options.
Source Files