worker

Documentation

Index

• func InitConfig(filePath string) (err error)
• func InitExecutor() (err error)
• func InitJobManage() (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() (err error)
• type JobManage
  • func (jobManage *JobManage) CreateJobLock(jobName string) (jobLock *JobLock)
• type LogSink
  • func (logSink *LogSink) Append(logRecord *common.JobLogRecord)
  • func (logSink *LogSink) WriteLoop()
• type Register
  • func (register *Register) KeepAlive()
• type Scheduler
  • func (scheduler *Scheduler) PushJobEvent(jobEvent *common.JobEvent)
  • func (scheduler *Scheduler) PushJobResult(result *common.JobExecuteResult)
  • func (scheduler *Scheduler) TryScheduler() (schedulerAfter time.Duration)
  • func (scheduler *Scheduler) TryStartJob(schedulerPlan *common.SchedulerPlan)

Functions

func InitConfig

func InitConfig(filePath string) (err error)

初始化配置信息

func InitExecutor

func InitExecutor() (err error)

初始化

func InitJobManage

func InitJobManage() (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 {
	// ==etcd配置
	// 集群
	EtcdEndPoints []string `json:"etcdEndPoints"`
	// etcd连接超时
	EtcdDialTimeout int `json:"etcdDialTimeout"`
}

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 {
	// 任务名称
	JobName string
	// etcd kv操作资源
	ETCDkv clientv3.KV
	// etcd 租约
	ETCDLease clientv3.Lease
	// 用于终止自动续租
	CancelFunc context.CancelFunc
	// 租约id
	LeaseId clientv3.LeaseID
	// 是否上锁成功
	IsLocked bool
}

分布式 任务锁（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() (err error)

任务释放锁

type JobManage

type JobManage struct {
	// etcd 客户端
	ETCDClient *clientv3.Client
	// etcd kv操作资源
	ETCDkv clientv3.KV
	// etcd 租约
	ETCDLease clientv3.Lease
	// etcd watcher
	ETCDWatcher clientv3.Watcher
}

任务管理

var (
	G_jobManage *JobManage
)

单例

func (*JobManage) CreateJobLock

func (jobManage *JobManage) 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(logRecord *common.JobLogRecord)

发送日志

func (*LogSink) WriteLoop

func (logSink *LogSink) WriteLoop()

日志存储协程

type Register

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

服务注册 /cron/works/{localIp}

var (
	G_register *Register
)

func (*Register) KeepAlive

func (register *Register) KeepAlive()

注册服务 注册到/cron/works/{localIp},并自动续租

type Scheduler

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

调度器

var (
	G_scheduler *Scheduler
)

func (*Scheduler) PushJobEvent

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

worker.watcher从etcd中获取任务变化 推送job事件到 scheduler

func (*Scheduler) PushJobResult

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

executor 执行完job 推送任务结果到 scheduler

func (*Scheduler) TryScheduler

func (scheduler *Scheduler) TryScheduler() (schedulerAfter time.Duration)

计算任务调度状态（扫描所有任务）

func (*Scheduler) TryStartJob

func (scheduler *Scheduler) TryStartJob(schedulerPlan *common.SchedulerPlan)

执行任务