dcron

README

dcron

分布式定时任务库

原理

基于redis同步节点数据，模拟服务注册。然后将任务名 根据一致性hash 选举出执行该任务的节点。

流程图

特性

• 负载均衡：根据任务数据和节点数据均衡分发任务。
• 无缝扩容：如果任务节点负载过大，直接启动新的服务器后部分任务会自动迁移至新服务实现无缝扩容。
• 故障转移：单个节点故障，10s后会自动将任务自动转移至其他正常节点。
• 任务唯一：同一个服务内同一个任务只会启动单个运行实例，不会重复执行。
• 自定义存储：通过实现driver接口来增加节点数据存储方式。

使用说明

1.创建redisDriver实例，指定服务名并初始化dcron。服务名为执行相同任务的单元。

  drv, _ := redis.NewDriver(&redis.Conf{
  		Host: "127.0.0.1",
  		Port: 6379,
  })
  dcron := NewDcron("server1", drv)

2.使用cron语法添加任务，需要指定任务名。任务名作为任务的唯一标识，必须保证唯一。

    dcron.AddFunc("test1","*/3 * * * *",func(){
		fmt.Println("执行 test1 任务",time.Now().Format("15:04:05"))
	})

3.开始任务。

dcron.Start()

关于服务名的说明

服务名只是为了定义相同一组任务，节点在启动时会产生一个uuid，然后绑定到这个服务内，不会存在多个节点使用同一个服务明出现冲突的问题。

比如有个服务叫【课堂服务】里面包含了 【上课】【下课】 等各类定时任务，那么就可以有n个不同的服务节点（可以在同一台或者不同机器上），服务都叫课堂服务。

其他

注意：一般定时如果和http服务在一起时不用特殊处理；但如果程序内只有该定时任务，需要阻塞主进程以防止主线程直接退出。

Documentation

Index

• type Dcron
  • func NewDcron(serverName string, driver Driver, cronOpts ...cron.Option) *Dcron
  • func NewDcronWithOption(serverName string, driver Driver, dcronOpts ...Option) *Dcron
  • func (d *Dcron) AddFunc(jobName, cronStr string, cmd func()) (err error)
  • func (d *Dcron) AddJob(jobName, cronStr string, job Job) (err error)
  • func (d *Dcron) GetLogger() *log.Logger
  • func (d *Dcron) Remove(jobName string)
  • func (d *Dcron) Run()
  • func (d *Dcron) SetLogger(logger *log.Logger)
  • func (d *Dcron) Start()
  • func (d *Dcron) Stop()
• type Job
• type JobWarpper
  • func (job JobWarpper) Run()
• type NodePool
  • func (np *NodePool) PickNodeByJobName(jobName string) string
  • func (np *NodePool) StartPool() error
• type Option
  • func CronOptionChain(wrappers ...cron.JobWrapper) Option
  • func CronOptionLocation(loc *time.Location) Option
  • func CronOptionParser(p cron.ScheduleParser) Option
  • func CronOptionSeconds() Option
  • func WithHashReplicas(d int) Option
  • func WithLogger(logger *log.Logger) Option
  • func WithNodeUpdateDuration(d time.Duration) Option

Types

type Dcron

type Dcron struct {
	ServerName string
	// contains filtered or unexported fields
}

Dcron is main struct

func NewDcron

func NewDcron(serverName string, driver Driver, cronOpts ...cron.Option) *Dcron

NewDcron create a Dcron

func NewDcronWithOption

func NewDcronWithOption(serverName string, driver Driver, dcronOpts ...Option) *Dcron

NewDcronWithOption create a Dcron with Dcron Option

func (*Dcron) AddFunc

func (d *Dcron) AddFunc(jobName, cronStr string, cmd func()) (err error)

AddFunc add a cron func

func (*Dcron) AddJob

func (d *Dcron) AddJob(jobName, cronStr string, job Job) (err error)

AddJob add a job

func (*Dcron) GetLogger

func (d *Dcron) GetLogger() *log.Logger

GetLogger get dcron logger

func (*Dcron) Remove

func (d *Dcron) Remove(jobName string)

Remove Job

func (*Dcron) Run

func (d *Dcron) Run()

Run Job

func (*Dcron) SetLogger

func (d *Dcron) SetLogger(logger *log.Logger)

SetLogger set dcron logger

func (*Dcron) Start

func (d *Dcron) Start()

Start start job

func (*Dcron) Stop

func (d *Dcron) Stop()

Stop stop job

type Job

type Job interface {
	Run()
}

type JobWarpper

type JobWarpper struct {
	ID      cron.EntryID
	Dcron   *Dcron
	Name    string
	CronStr string
	Func    func()
	Job     Job
}

JobWarpper is a job warpper

func (JobWarpper) Run

func (job JobWarpper) Run()

Run is run job

type NodePool

type NodePool struct {
	NodeID string

	Driver driver.Driver
	// contains filtered or unexported fields
}

NodePool is a node pool

func (*NodePool) PickNodeByJobName

func (np *NodePool) PickNodeByJobName(jobName string) string

PickNodeByJobName : 使用一致性hash算法根据任务名获取一个执行节点

func (*NodePool) StartPool

func (np *NodePool) StartPool() error

type Option

type Option func(*Dcron)

func CronOptionChain

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

CronOptionChain is Warp cron with chain

func CronOptionLocation

func CronOptionLocation(loc *time.Location) Option

CronOptionLocation is warp cron with location

func CronOptionParser

func CronOptionParser(p cron.ScheduleParser) Option

CronOptionParser is warp cron with schedules.

func CronOptionSeconds

func CronOptionSeconds() Option

CronOptionSeconds is warp cron with seconds

func WithHashReplicas

func WithHashReplicas(d int) Option

WithHashReplicas set hashReplicas

func WithLogger

func WithLogger(logger *log.Logger) Option

WithLogger both set dcron and cron logger.

func WithNodeUpdateDuration

func WithNodeUpdateDuration(d time.Duration) Option

WithNodeUpdateDuration set node update duration