cronlib

README

cronlib

cronlib基于github.com/go-co-op/gocron的基础上，添加了redis分布式锁，并简化了定时任务调度器的使用；

基于分布式锁的操作，用以解决在集群模式下，多cron可能在同一时刻发起调度的问题

cronlib 快速使用流程

1. 初始化Job业务逻辑：定义Job定期执行具体业务逻辑
2. 初始化Job任务: NewJob()方法创建调度job
3. 初始化Scheduler调度器
4. 添加Job任务到调度器
5. 调度器启动

简单的定时任务示例

func ExampleStartTest() {
	var job1Fn = func(jobID int) {
		log.Printf("exec job[#%d] fn", jobID)
	}

	// 准备job任务
	job1 := NewJob("job1", func() { job1Fn(1) }, "* * * * *")
	job2 := NewJob("job2", func() { job1Fn(2) }, "*/2 * * * *")

	// 准备scheduler调度器
	locker, err := NewRedisLocker("redis://127.0.0.1:6379")
	if err != nil {
		log.Fatalf("parse redis url got err: %s", err)
	}

	crond := NewScheduler(&Config{
		Async:            false,           // 不阻塞主协程
		SingletonModeAll: true,            // 调度器不会重复调度同类型新的job任务
	}, locker)

	// 添加job任务
	err = crond.AddJobs(job1, job2)
	if err != nil {
		log.Fatalf("crond add job got err: %s", err)
	}

	// job执行
	crond.Start()
}

redis 依赖

通过docker快速启动一个本地redis实例：

$ docker-compose up -d

Documentation

Overview

Package cron is a generated GoMock package.

Index

• Constants
• func SetDebug(b bool)
• type Config
• type Job
  • func NewJob(name string, exec func(), express string) *Job
• type Locker
• type MockLocker
  • func NewMockLocker(ctrl *gomock.Controller) *MockLocker
  • func (m *MockLocker) EXPECT() *MockLockerMockRecorder
  • func (m *MockLocker) Lock(key string, ttl time.Duration) error
  • func (m *MockLocker) UnLock(key string) error
• type MockLockerMockRecorder
  • func (mr *MockLockerMockRecorder) Lock(key, ttl interface{}) *gomock.Call
  • func (mr *MockLockerMockRecorder) UnLock(key interface{}) *gomock.Call
• type RedisLocker
  • func NewRedisLocker(redisURL string) (*RedisLocker, error)
  • func (r *RedisLocker) Lock(key string, ttl time.Duration) error
  • func (r *RedisLocker) UnLock(key string) error
• type Scheduler
  • func NewScheduler(cfg *Config, locker Locker) *Scheduler
  • func (s *Scheduler) AddJobs(jobs ...*Job) error
  • func (s *Scheduler) Start()

Constants

const (
	ExpressEveryMin   = "* * * * *"
	ExpressEvery5Min  = "*/5 * * * *"
	ExpressEvery10Min = "*/10 * * * *"
	ExpressEvery30Min = "*/30 * * * *"
	ExpressEveryHour  = "0 * * * *" // 每小时整点执行
	ExpressEveryDay   = "0 1 * * *" // 凌晨1:00执行
	ExpressEveryMonth = "0 1 1 * *" // 每个月
)

cron格式`*/1 * * * *`，分(0-59) 时(0-23) 日(1-31) 月(1-12) 天(0-6)

Functions

func SetDebug

func SetDebug(b bool)

Types

type Config

type Config struct {
	TimeZone         string // 时区配置，默认为Asia/Shanghai
	Async            bool   // 启动job的方式是阻塞还是非阻塞
	SingletonModeAll bool   // 启动job是否采用单例模式，单例模式下若job如果之前有运行且未完成，则调度器不会重复调度同类型新的job任务(若无特殊要求，推荐开启)
}

Config Cron调度器配置

type Job

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

Job 待执行的JOB任务

func NewJob

func NewJob(name string, exec func(), express string) *Job

NewJob 创建一个任务

type Locker

type Locker interface {
	// Lock 分布式加锁
	Lock(key string, ttl time.Duration) error

	// UnLock 分布式解锁
	UnLock(key string) error
}

Locker 分布式锁接口 该接口主要是针对多个JOB执行时刻，进行并发控制，确保在不同机器上的Job仅执行一次行

type MockLocker

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

MockLocker is a mock of Locker interface

func NewMockLocker

func NewMockLocker(ctrl *gomock.Controller) *MockLocker

NewMockLocker creates a new mock instance

func (*MockLocker) EXPECT

func (m *MockLocker) EXPECT() *MockLockerMockRecorder

EXPECT returns an object that allows the caller to indicate expected use

func (*MockLocker) Lock

func (m *MockLocker) Lock(key string, ttl time.Duration) error

Lock mocks base method

func (*MockLocker) UnLock

func (m *MockLocker) UnLock(key string) error

UnLock mocks base method

type MockLockerMockRecorder

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

MockLockerMockRecorder is the mock recorder for MockLocker

func (*MockLockerMockRecorder) Lock

func (mr *MockLockerMockRecorder) Lock(key, ttl interface{}) *gomock.Call

Lock indicates an expected call of Lock

func (*MockLockerMockRecorder) UnLock

func (mr *MockLockerMockRecorder) UnLock(key interface{}) *gomock.Call

UnLock indicates an expected call of UnLock

type RedisLocker

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

RedisLocker 基于Redis实现的分布式锁，

func NewRedisLocker

func NewRedisLocker(redisURL string) (*RedisLocker, error)

NewRedisLocker 创建一个Redis分布式锁 Examples:

redis://user:password@localhost:6789/3?dial_timeout=3&db=1&read_timeout=6s&max_retries=2
is equivalent to:
&Options{
	Network:     "tcp",
	Addr:        "localhost:6789",
	DB:          1,               // path "/3" was overridden by "&db=1"
	DialTimeout: 3 * time.Second, // no time unit = seconds
	ReadTimeout: 6 * time.Second,
	MaxRetries:  2,
}

func (*RedisLocker) Lock

func (r *RedisLocker) Lock(key string, ttl time.Duration) error

Lock 分布式加锁

func (*RedisLocker) UnLock

func (r *RedisLocker) UnLock(key string) error

UnLock 分布式解锁

type Scheduler

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

Scheduler Cron调度程序

func NewScheduler

func NewScheduler(cfg *Config, locker Locker) *Scheduler

NewScheduler 初始化一个cron 调度器

func (*Scheduler) AddJobs

func (s *Scheduler) AddJobs(jobs ...*Job) error

AddJobs 新增一个cron job任务

func (*Scheduler) Start

func (s *Scheduler) Start()

Start 启动Cron定时服务