dcron

README

dcron

分布式定时任务库

原理

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

申明

本代码参考 LibiChai/dcron,但是在使用过程中，原代码封装行不是很良好，健壮很差，不能用于生产环境，对代码结构进行部分重构，但是改变太大， 暂时放弃pr到原库

流程图

特性

• 负载均衡：根据任务数据和节点数据均衡分发任务。
• 无缝扩容：如果任务节点负载过大，直接启动新的服务器后部分任务会自动迁移至新服务实现无缝扩容。
• 故障转移：单个节点故障，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, opts ...cron.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) Remove(jobName string)
  • func (d *Dcron) Run()
  • func (d *Dcron) Start()
  • func (d *Dcron) Stop()
• type Job
• type JobWarpper
  • func (job JobWarpper) Run()
• type NodePool
  • func (np *NodePool) InitPoolGrabService()
  • func (np *NodePool) PickNodeByJobName(jobName string) string
  • func (np *NodePool) SetInterval(interval time.Duration)
• type PoolOptions

Types

type Dcron

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

Dcron is main struct

func NewDcron

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

NewDcron create a Dcron

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) Remove

func (d *Dcron) Remove(jobName string)

Remove Job

func (*Dcron) Run

func (d *Dcron) Run()

Run Job

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) InitPoolGrabService

func (np *NodePool) InitPoolGrabService()

func (*NodePool) PickNodeByJobName

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

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

func (*NodePool) SetInterval

func (np *NodePool) SetInterval(interval time.Duration)

SetInterval set interval time

type PoolOptions

type PoolOptions struct {
	Replicas int
	HashFn   consistenthash.Hash
}

PoolOptions is a pool options