worker

Documentation

Index

• func InitConfig(filename string) (err error)
• func InitExecutor() (err error)
• func InitJobMgr() (err error)
• func InitLogSink() (err error)
• func InitRegister() (err error)
• func InitScheduler() (err error)
• type Config
• type Executor
  • func (executor *Executor) ExecuteJob(info *common.JobExecuteInfo)
• type JobLock
  • func InitJobLock(jobName string, kv clientv3.KV, lease clientv3.Lease) (jobLock *JobLock)
  • func (jobLock *JobLock) TryLock() (err error)
  • func (jobLock *JobLock) Unlock()
• type JobMgr
  • func (jobMgr *JobMgr) CreateJobLock(jobName string) (jobLock *JobLock)
• type LogSink
  • func (logSink *LogSink) Append(jobLog *common.JobLog)
• type Register
• type Scheduler
  • func (scheduler *Scheduler) PushJobEvent(jobEvent *common.JobEvent)
  • func (scheduler *Scheduler) PushJobResult(jobResult *common.JobExecuteResult)
  • func (scheduler *Scheduler) TrySchedule() (scheduleAfter time.Duration)
  • func (scheduler *Scheduler) TryStartJob(jobPlan *common.JobSchedulePlan)

Functions

func InitConfig

func InitConfig(filename string) (err error)

func InitExecutor

func InitExecutor() (err error)

初始化执行器

func InitJobMgr

func InitJobMgr() (err error)

func InitLogSink

func InitLogSink() (err error)

func InitRegister

func InitRegister() (err error)

func InitScheduler

func InitScheduler() (err error)

初始化调度器

Types

type Config

type Config struct {
	EtcdEndPoints         []string `json:"etcdEndPoints"`   //etcd集群列表
	EtcdDialTimeout       int      `json:"etcdDialTimeout"` //etcd超时时间
	MongodbUri            string   `json:"mongodbUri"`
	MongodbConnectTimeout int      `json:"mongodbConnectTimeout"`
	JobLogBatchSize       int      `json:"jobLogBatchSize"`
	JobLogCommitTimeout   int      `json:"jobLogCommitTimeout"`
}

配置列表

var (
	G_config *Config
)

type Executor

type Executor struct {
}

任务执行器

var (
	G_executor *Executor
)

func (*Executor) ExecuteJob

func (executor *Executor) ExecuteJob(info *common.JobExecuteInfo)

执行一个任务

type JobLock

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

分布式锁（通过抢占一个TXN事务，谁先抢到谁就占到了🔐）

func InitJobLock

func InitJobLock(jobName string, kv clientv3.KV, lease clientv3.Lease) (jobLock *JobLock)

初始化一把🔐

func (*JobLock) TryLock

func (jobLock *JobLock) TryLock() (err error)

尝试上锁

func (*JobLock) Unlock

func (jobLock *JobLock) Unlock()

释放锁

type JobMgr

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

var (
	G_jobMgr *JobMgr
)

func (*JobMgr) CreateJobLock

func (jobMgr *JobMgr) CreateJobLock(jobName string) (jobLock *JobLock)

创建任务执行锁

type LogSink

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

mongodb存储日志

var (
	G_logSink *LogSink
)

func (*LogSink) Append

func (logSink *LogSink) Append(jobLog *common.JobLog)

发送日志

type Register

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

注册节点到etcd：/cron/workers/IP地址

var (
	G_register *Register
)

type Scheduler

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

任务调度

var (
	G_scheduler *Scheduler
)

func (*Scheduler) PushJobEvent

func (scheduler *Scheduler) PushJobEvent(jobEvent *common.JobEvent)

推送任务变化事件

func (*Scheduler) PushJobResult

func (scheduler *Scheduler) PushJobResult(jobResult *common.JobExecuteResult)

回传任务执行结果

func (*Scheduler) TrySchedule

func (scheduler *Scheduler) TrySchedule() (scheduleAfter time.Duration)

重新计算任务调度状态

func (*Scheduler) TryStartJob

func (scheduler *Scheduler) TryStartJob(jobPlan *common.JobSchedulePlan)

尝试执行任务