gxxljob

Documentation

Index

• Constants
• type Executor
  • func Init(opts Options) *Executor
  • func RunWithDefaultOptions(appName string) (executor *Executor, err error)
  • func RunWithDefaultOptionsLogger(logger *slog.Logger) (executor *Executor, err error)
  • func RunWithDefaultOptionsLoggerAndAppName(appName string, logger *slog.Logger) (executor *Executor, err error)
  • func RunWithOptions(opts Options) (executor *Executor, err error)
  • func (e *Executor) RegTask(pattern string, task TaskFunc)
  • func (e *Executor) Run() (err error)
  • func (e *Executor) Stop()
• type IdleBeatReq
• type JobHandleResult
• type KillRequest
• type Options
  • func (opt *Options) WithAccessToken(accessToken string) *Options
  • func (opt *Options) WithAdminAddresses(adminAddresses string) *Options
  • func (opt *Options) WithAppName(appName string) *Options
  • func (opt *Options) WithExecutorAddress(executorAddress string) *Options
  • func (opt *Options) WithExecutorPort(executorPort int64) *Options
  • func (opt *Options) WithLogPath(logPath string) *Options
  • func (opt *Options) WithLogger(logger *slog.Logger) *Options
  • func (opt *Options) WithLogretentiondays(logretentiondays uint64) *Options
• type Registry
• type Resp
• type RunLogRequest
• type RunLogRespContent
• type RunRequest
• type Task
  • func (t *Task) Info() string
  • func (t *Task) Run(callback func(code int64, msg string))
• type TaskFunc

Constants

const (
	SuccessCode = 200
	FailureCode = 500

	DefaultExecutorPort = 9999

	DefaultRegistryGroup = "EXECUTOR"

	DefaultAppName = "go-xxljob"

	DefaultRegisterAddressHttp = "http://"
)

const (
	AccessTokenHeaderKey = "XXL-JOB-ACCESS-TOKEN"
	DefaultBeatInterval  = 20
)

Types

type Executor

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

Executor 执行器

func Init

func Init(opts Options) *Executor

func RunWithDefaultOptions

func RunWithDefaultOptions(appName string) (executor *Executor, err error)

func RunWithDefaultOptionsLogger

func RunWithDefaultOptionsLogger(logger *slog.Logger) (executor *Executor, err error)

func RunWithDefaultOptionsLoggerAndAppName

func RunWithDefaultOptionsLoggerAndAppName(appName string, logger *slog.Logger) (executor *Executor, err error)

func RunWithOptions

func RunWithOptions(opts Options) (executor *Executor, err error)

func (*Executor) RegTask

func (e *Executor) RegTask(pattern string, task TaskFunc)

RegTask 注册任务

func (*Executor) Run

func (e *Executor) Run() (err error)

func (*Executor) Stop

func (e *Executor) Stop()

type IdleBeatReq

type IdleBeatReq struct {
	JobId int64 `json:"jobId"` // 任务ID
}

忙碌检测:调度中心检测指定执行器上指定任务是否忙碌（运行中）时使用

type JobHandleResult

type JobHandleResult struct {
	LogID      int64  `json:"logId"`      // 本次调度日志ID
	LogDateTim int64  `json:"logDateTim"` // 本次调度日志时间
	HandleCode int64  `json:"handleCode"` //200表示正常,500表示失败
	HandleMsg  string `json:"handleMsg"`
}

type KillRequest

type KillRequest struct {
	JobID int64 `json:"jobId"` // 任务ID
}

说明：终止任务

type Options

type Options struct {
	// 调度中心部署根地址 [选填]：如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调"；为空则关闭自动注册
	AdminAddresses string `json:"admin_addresses" property:"xxl.job.admin.addresses"`

	//执行器通讯TOKEN [选填]：非空时启用；
	AccessToken string `json:"access_token" property:"xxl.job.accessToken"`
	//执行器AppName [选填]：执行器心跳注册分组依据；为空则关闭自动注册
	AppName string        `json:"app_name" property:"xxl.job.executor.appname"`
	Timeout time.Duration `json:"timeout" property:"xxl.job.executor.timeout"` //接口超时时间
	//执行器注册 [选填]：优先使用该配置作为注册地址，为空时使用内嵌服务 ”IP:PORT“ 作为注册地址。从而更灵活的支持容器类型执行器动态IP和动态映射端口问题
	ExecutorAddress string `json:"executor_address" property:"xxl.job.executor.address"`
	//执行器端口号 [选填]：小于等于0则自动获取；默认端口为9999，单机部署多个执行器时，注意要配置不同执行器端口；
	ExecutorPort int64 `json:"executor_port" property:"xxl.job.executor.port"`
	Logger       *slog.Logger
	// 执行器运行日志文件存储磁盘路径 [选填] ：需要对该路径拥有读写权限；为空则使用默认路径；
	LogPath string `json:"log_path" property:"xxl.job.executor.logpath"`
	//执行器日志文件保存天数 [选填] ： 过期日志自动清理, 限制值大于等于3时生效; 否则, 如-1, 关闭自动清理功能；
	Logretentiondays uint64 `json:"logretentiondays" property:"xxl.job.executor.logretentiondays"`
	LogJsonFormat    bool   `json:"logJsonFormat" property:"xxl.job.executor.logJsonFormat"`
	//执行器心跳间隔单位秒
	BeatInterval uint64 `json:"beatInterval" property:"xxl.job.executor.beatInterval"`
	// contains filtered or unexported fields
}

func (*Options) WithAccessToken

func (opt *Options) WithAccessToken(accessToken string) *Options

func (*Options) WithAdminAddresses

func (opt *Options) WithAdminAddresses(adminAddresses string) *Options

func (*Options) WithAppName

func (opt *Options) WithAppName(appName string) *Options

func (*Options) WithExecutorAddress

func (opt *Options) WithExecutorAddress(executorAddress string) *Options

func (*Options) WithExecutorPort

func (opt *Options) WithExecutorPort(executorPort int64) *Options

func (*Options) WithLogPath

func (opt *Options) WithLogPath(logPath string) *Options

func (*Options) WithLogger

func (opt *Options) WithLogger(logger *slog.Logger) *Options

func (*Options) WithLogretentiondays

func (opt *Options) WithLogretentiondays(logretentiondays uint64) *Options

type Registry

type Registry struct {
	RegistryGroup string `json:"registryGroup"` // 固定值EXECUTOR
	RegistryKey   string `json:"registryKey"`   // 执行器AppName
	RegistryValue string `json:"registryValue"` // 执行器地址，内置服务跟地址
}

type Resp

type Resp struct {
	Code    int64       `json:"code"` // 200 表示正常、其他失败
	Msg     interface{} `json:"msg"`  // 错误提示消息
	Content any         `json:"content"`
}

接口响应结果

type RunLogRequest

type RunLogRequest struct {
	LogID       int64 `json:"logId"`       // 本次调度日志ID
	LogDateTime int64 `json:"logDateTime"` // 本次调度日志时间
	FromLineNum int64 `json:"fromLineNum"` // 日志开始行号，滚动加载日志
}

type RunLogRespContent

type RunLogRespContent struct {
	FromLineNum int64  `json:"fromLineNum"` // 本次请求，日志开始行数
	ToLineNum   int64  `json:"toLineNum"`   // 本次请求，日志结束行号
	LogContent  string `json:"logContent"`  // 本次请求日志内容
	IsEnd       bool   `json:"isEnd"`       // 日志是否全部加载完
}

LogResContent 日志响应内容

type RunRequest

type RunRequest struct {
	JobID                 int64  `json:"jobId"`                 // 任务ID
	ExecutorHandler       string `json:"executorHandler"`       // 任务标识
	ExecutorParams        string `json:"executorParams"`        // 任务参数
	ExecutorBlockStrategy string `json:"executorBlockStrategy"` // 任务阻塞策略
	ExecutorTimeout       int64  `json:"executorTimeout"`       // 任务超时时间，单位秒，大于零时生效
	LogID                 int64  `json:"logId"`                 // 本次调度日志ID
	LogDateTime           int64  `json:"logDateTime"`           // 本次调度日志时间
	GlueType              string `json:"glueType"`              // 任务模式，可选值参考 com.xxl.job.core.glue.GlueTypeEnum
	GlueSource            string `json:"glueSource"`            // GLUE脚本代码
	GlueUpdatetime        int64  `json:"glueUpdatetime"`        // GLUE脚本更新时间，用于判定脚本是否变更以及是否需要刷新
	BroadcastIndex        int64  `json:"broadcastIndex"`        // 分片参数：当前分片
	BroadcastTotal        int64  `json:"broadcastTotal"`        // 分片参数：总分片
}

说明：触发任务执行

type Task

type Task struct {
	Id    int64
	Name  string
	Ctx   context.Context
	Param *RunRequest

	Cancel    context.CancelFunc
	StartTime int64
	EndTime   int64
	// contains filtered or unexported fields
}

Task 任务

func (*Task) Info

func (t *Task) Info() string

Info 任务信息

func (*Task) Run

func (t *Task) Run(callback func(code int64, msg string))

Run 运行任务

type TaskFunc

type TaskFunc func(cxt context.Context, param *RunRequest) error

TaskFunc 任务执行函数