xxl

README

forked from forked from https://github.com/ainiaa/xxl-job-executor-go 进行了一些调整：

• 执行器执行完成后，调用 callback 方法：如果 xxl-job admin 已经断开，则会导致本应用 panic
• 执行 returnCall 方法，向xxl-job admin 提交的数据，其结构与 xxl-job 所要求的不符（可能是xxl-job已经更新的缘故）
• 心跳 log 太啰嗦，简化
• server.ListenAndServe 没有处理异常，导致明明启动失败，确看起来一切正常
• 给Task增加Description

xxl-job-executor-go

很多公司java与go开发共存，java中有xxl-job做为任务调度引擎，为此也出现了go执行器(客户端)，使用起来比较简单：

支持

1.执行器注册
2.耗时任务取消
3.任务注册，像写http.Handler一样方便
4.任务panic处理
5.阻塞策略处理
6.任务完成支持返回执行备注
7.任务超时取消 (单位：秒，0为不限制)
8.失败重试次数(在参数param中，目前由任务自行处理)
9.可自定义日志
10.自定义日志查看handler
11.支持外部路由（可与gin集成）

Example

import (
	"fmt"
	xxl "github.com/ainiaa/xxl-job-executor-go"
	"github.com/ainiaa/xxl-job-executor-go/example/task"
	"log"
)

// Option style
func initViaOption() xxl.Executor{
	return xxl.NewExecutor(
		xxl.ServerAddr("http://127.0.0.1/xxl-job-admin"),
		xxl.AccessToken(""),            //请求令牌(默认为空)
		xxl.ExecutorIp("127.0.0.1"),    //可自动获取
		xxl.ExecutorPort("9999"),       //默认9999（非必填）
		xxl.RegistryKey("golang-jobs"), //执行器名称
		xxl.SetLogger(&logger{}),       //自定义日志
	)
}
// Config style
func initViaConfig() xxl.Executor{
	c := xxl.Conf{
		ServerAddr:   "http://127.0.0.1/xxl-job-admin",
		ExecutorPort: "9999",
		RegistryKey:  "golang-jobs",
		AccessToken:  "",
	}
	return xxl.New(c)
}

func main() {
	exec := initViaConfig()
	exec.SetLogger(&logger{})
	exec.Init()
	//设置日志查看handler
	exec.LogHandler(func(req *xxl.LogReq) *xxl.LogRes {
		return &xxl.LogRes{Code: 200, Msg: "", Content: xxl.LogResContent{
			FromLineNum: req.FromLineNum,
			ToLineNum:   2,
			LogContent:  "这个是自定义日志handler",
			IsEnd:       true,
		}}
	})
	//注册任务handler
	exec.RegTask("task.test", task.Test)
	exec.RegTask("task.test2", task.Test2)
	exec.RegTask("task.panic", task.Panic)
	log.Fatal(exec.Run())
}

//xxl.Logger接口实现
type logger struct{}

func (l *logger) Fatalf(format string, a ...interface{}) {
	panic("implement me")
}

func (l *logger) Infof(format string, a ...interface{}) {
	fmt.Println(fmt.Sprintf("自定义日志 - "+format, a...))
}

func (l *logger) Errorf(format string, a ...interface{}) {
	log.Println(fmt.Sprintf("自定义日志 - "+format, a...))
}

示例项目

github.com/ainiaa/xxl-job-executor-go/example/

与gin框架集成

https://github.com/ainiaa/gin-xxl-job-executor

xxl-job-admin配置

添加执行器

执行器管理->新增执行器,执行器列表如下：

AppName		名称		注册方式	OnLine 		机器地址 		操作
golang-jobs	golang执行器	自动注册 		查看 ( 1 ）   

查看->注册节点

http://127.0.0.1:9999

添加任务

任务管理->新增(注意，使用BEAN模式，JobHandler与RegTask名称一致)

1	测试panic	BEAN：task.panic	* 0 * * * ?	admin	STOP	
2	测试耗时任务	BEAN：task.test2	* * * * * ?	admin	STOP	
3	测试golang	BEAN：task.test		* * * * * ?	admin	STOP

Documentation

Index

• Constants
• Variables
• func Int64ToStr(i int64) string
• type Conf
• type ExecuteResult
• type Executor
  • func New(c Conf) Executor
  • func NewExecutor(opts ...Option) Executor
• type LogFunc
• type LogHandler
• type LogReq
• type LogRes
• type LogResContent
• type Logger
• type Option
  • func AccessToken(token string) Option
  • func ExecutorIp(ip string) Option
  • func ExecutorPort(port string) Option
  • func RegistryKey(registryKey string) Option
  • func ServerAddr(addr string) Option
  • func SetLogger(l Logger) Option
• type Options
• type Registry
• type RunReq
• 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
)

响应码

const (
	MESSAGE_SUCCESS = "ok"
)

Variables

var (
	DefaultExecutorPort = "9999"
	DefaultRegistryKey  = "golang-jobs"
)

Functions

func Int64ToStr

func Int64ToStr(i int64) string

Int64ToStr int64 to str

Types

type Conf

type Conf struct {
	ServerAddr   string `json:"server_addr"`
	ExecutorPort string `json:"executor_port"`
	RegistryKey  string `json:"registry_key"`
	AccessToken  string `json:"access_token"`
}

type ExecuteResult

type ExecuteResult struct {
	Code int64       `json:"code"`
	Msg  interface{} `json:"msg"`
}

ExecuteResult 任务执行结果 200 表示任务执行正常，500表示失败

type Executor

type Executor interface {
	// Init 初始化
	Init(...Option)
	// LogHandler 日志查询
	LogHandler(handler LogHandler)
	// RegTask 注册任务
	RegTask(pattern string, desc string, task TaskFunc)
	// RunTask 运行任务
	RunTask(writer http.ResponseWriter, request *http.Request)
	// KillTask 杀死任务
	KillTask(writer http.ResponseWriter, request *http.Request)
	// TaskLog 任务日志
	TaskLog(writer http.ResponseWriter, request *http.Request)
	// Beat 心跳检测
	Beat(writer http.ResponseWriter, request *http.Request)
	// IdleBeat 忙碌检测
	IdleBeat(writer http.ResponseWriter, request *http.Request)
	// Run 运行服务
	Run() error
	// Stop 停止服务
	Stop()
	// SetLogger logger
	SetLogger(log Logger)
	// RegistryRemove 摘除 xxljob
	RegistryRemove()
	// GetAllTasks 显示所有已经注册的任务
	GetAllTasks() map[string]*Task
	// ExpireLog 自动清理日志
	ExpireLog(days int, dir string)
}

Executor 执行器

func New

func New(c Conf) Executor

创建执行器

func NewExecutor

func NewExecutor(opts ...Option) Executor

NewExecutor 创建执行器

type LogFunc

type LogFunc func(req LogReq, res *LogRes) []byte

LogFunc 应用日志

type LogHandler

type LogHandler func(req *LogReq) *LogRes

type LogReq

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

LogReq 日志请求

type LogRes

type LogRes struct {
	Code    int64         `json:"code"`    // 200 表示正常、其他失败
	Msg     string        `json:"msg"`     // 错误提示消息
	Content LogResContent `json:"content"` // 日志响应内容
}

LogRes 日志响应

type LogResContent

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

LogResContent 日志响应内容

type Logger

type Logger interface {
	Info(format string, a ...interface{})
	Error(format string, a ...interface{})
}

Logger 系统日志

type Option

type Option func(o *Options)

func AccessToken

func AccessToken(token string) Option

AccessToken 请求令牌

func ExecutorIp

func ExecutorIp(ip string) Option

ExecutorIp 设置执行器IP

func ExecutorPort

func ExecutorPort(port string) Option

ExecutorPort 设置执行器端口

func RegistryKey

func RegistryKey(registryKey string) Option

RegistryKey 设置执行器标识

func ServerAddr

func ServerAddr(addr string) Option

ServerAddr 设置调度中心地址

func SetLogger

func SetLogger(l Logger) Option

SetLogger 设置日志处理器

type Options

type Options struct {
	ServerAddr   string        `json:"server_addr"`   //调度中心地址
	AccessToken  string        `json:"access_token"`  //请求令牌
	Timeout      time.Duration `json:"timeout"`       //接口超时时间
	ExecutorIp   string        `json:"executor_ip"`   //本地(执行器)IP(可自行获取)
	ExecutorPort string        `json:"executor_port"` //本地(执行器)端口
	RegistryKey  string        `json:"registry_key"`  //执行器名称
	LogDir       string        `json:"log_dir"`       //日志目录
	// contains filtered or unexported fields
}

type Registry

type Registry struct {
	RegistryGroup string `json:"registryGroup"`
	RegistryKey   string `json:"registryKey"`
	RegistryValue string `json:"registryValue"`
}

Registry 注册参数

type RunReq

type RunReq 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"`        // 分片参数：总分片
}

RunReq 触发任务请求参数

type Task

type Task struct {
	Id    int64
	Name  string
	Ext   context.Context
	Param *RunReq

	Cancel    context.CancelFunc
	StartTime int64
	EndTime   int64
	Desc      string
	// 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 *RunReq) string

任务执行函数