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"`
	EtcdDialTimeout       int      `json:"etcdDialTimeout"`
	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)

尝试执行任务