modify log

constants.go

+package xxl
+
+// 响应码
+const (
+	SuccessCode = 200
+	FailureCode = 500
+)

dto.go

+package xxl
+
+//通用响应
+type res struct {
+	Code int64       `json:"code"` // 200 表示正常、其他失败
+	Msg  interface{} `json:"msg"`  // 错误提示消息
+}
+
+/*****************  上行参数  *********************/
+
+// Registry 注册参数
+type Registry struct {
+	RegistryGroup string `json:"registryGroup"`
+	RegistryKey   string `json:"registryKey"`
+	RegistryValue string `json:"registryValue"`
+}
+
+//执行器执行完任务后，回调任务结果时使用
+type call []*callElement
+
+type callElement struct {
+	LogID         int64          `json:"logId"`
+	LogDateTim    int64          `json:"logDateTim"`
+	ExecuteResult *ExecuteResult `json:"executeResult"`
+	//以下是7.31版本 v2.3.0 Release所使用的字段
+	HandleCode int    `json:"handleCode"` //200表示正常,500表示失败
+	HandleMsg  string `json:"handleMsg"`
+}
+
+// ExecuteResult 任务执行结果 200 表示任务执行正常，500表示失败
+type ExecuteResult struct {
+	Code int64       `json:"code"`
+	Msg  interface{} `json:"msg"`
+}
+
+/*****************  下行参数  *********************/
+
+//阻塞处理策略
+const (
+	serialExecution = "SERIAL_EXECUTION" //单机串行
+	discardLater    = "DISCARD_LATER"    //丢弃后续调度
+	coverEarly      = "COVER_EARLY"      //覆盖之前调度
+)
+
+// 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"`        // 分片参数：总分片
+}
+
+//终止任务请求参数
+type killReq struct {
+	JobID int64 `json:"jobId"` // 任务ID
+}
+
+//忙碌检测请求参数
+type idleBeatReq struct {
+	JobID int64 `json:"jobId"` // 任务ID
+}
+
+// LogReq 日志请求
+type LogReq struct {
+	LogDateTim  int64 `json:"logDateTim"`  // 本次调度日志时间
+	LogID       int64 `json:"logId"`       // 本次调度日志ID
+	FromLineNum int   `json:"fromLineNum"` // 日志开始行号，滚动加载日志
+}
+
+// LogRes 日志响应
+type LogRes struct {
+	Code    int64         `json:"code"`    // 200 表示正常、其他失败
+	Msg     string        `json:"msg"`     // 错误提示消息
+	Content LogResContent `json:"content"` // 日志响应内容
+}
+
+// LogResContent 日志响应内容
+type LogResContent struct {
+	FromLineNum int    `json:"fromLineNum"` // 本次请求，日志开始行数
+	ToLineNum   int    `json:"toLineNum"`   // 本次请求，日志结束行号
+	LogContent  string `json:"logContent"`  // 本次请求日志内容
+	IsEnd       bool   `json:"isEnd"`       // 日志是否全部加载完
+}

example/main.go

+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	xxl "gitlab.wanzhuangkj.com/tush/xxl-job-executor-go"
+	"gitlab.wanzhuangkj.com/tush/xxl-job-executor-go/example/task"
+)
+
+func main() {
+	exec := 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{}),       //自定义日志
+	)
+	exec.Init()
+	exec.Use(customMiddleware)
+	//设置日志查看handler
+	exec.LogHandler(customLogHandle)
+	//注册任务handler
+	exec.RegTask("task.test", task.Test)
+	exec.RegTask("task.test2", task.Test2)
+	exec.RegTask("task.panic", task.Panic)
+	log.Fatal(exec.Run())
+}
+
+// 自定义日志处理器
+func customLogHandle(req *xxl.LogReq) *xxl.LogRes {
+	return &xxl.LogRes{Code: xxl.SuccessCode, Msg: "", Content: xxl.LogResContent{
+		FromLineNum: req.FromLineNum,
+		ToLineNum:   2,
+		LogContent:  "这个是自定义日志handler",
+		IsEnd:       true,
+	}}
+}
+
+// xxl.Logger接口实现
+type logger struct{}
+
+func (l *logger) Info(format string, a ...interface{}) {
+	fmt.Println(fmt.Sprintf("自定义日志 - "+format, a...))
+}
+
+func (l *logger) Error(format string, a ...interface{}) {
+	log.Println(fmt.Sprintf("自定义日志 - "+format, a...))
+}
+
+// 自定义中间件
+func customMiddleware(tf xxl.TaskFunc) xxl.TaskFunc {
+	return func(cxt context.Context, param *xxl.RunReq) string {
+		log.Println("I am a middleware start")
+		res := tf(cxt, param)
+		log.Println("I am a middleware end")
+		return res
+	}
+}

example/task/panic.go

+package task
+
+import (
+	"context"
+
+	xxl "gitlab.wanzhuangkj.com/tush/xxl-job-executor-go"
+)
+
+func Panic(cxt context.Context, param *xxl.RunReq) (msg string) {
+	panic("test panic")
+	return
+}

example/task/test.go

+package task
+
+import (
+	"context"
+	"fmt"
+
+	xxl "gitlab.wanzhuangkj.com/tush/xxl-job-executor-go"
+)
+
+func Test(cxt context.Context, param *xxl.RunReq) (msg string) {
+	fmt.Println("test one task" + param.ExecutorHandler + " param：" + param.ExecutorParams + " log_id:" + xxl.Int64ToStr(param.LogID))
+	return "test done"
+}

example/task/test2.go

+package task
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	xxl "gitlab.wanzhuangkj.com/tush/xxl-job-executor-go"
+)
+
+func Test2(cxt context.Context, param *xxl.RunReq) (msg string) {
+	num := 1
+	for {
+
+		select {
+		case <-cxt.Done():
+			fmt.Println("task" + param.ExecutorHandler + "被手动终止")
+			return
+		default:
+			num++
+			time.Sleep(10 * time.Second)
+			fmt.Println("test one task"+param.ExecutorHandler+" param："+param.ExecutorParams+"执行行", num)
+			if num > 10 {
+				fmt.Println("test one task" + param.ExecutorHandler + " param：" + param.ExecutorParams + "执行完毕！")
+				return
+			}
+		}
+	}
+
+}

executor.go

+package xxl
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+)
+
+// Executor 执行器
+type Executor interface {
+	// Init 初始化
+	Init(...Option)
+	// LogHandler 日志查询
+	LogHandler(handler LogHandler)
+	// Use 使用中间件
+	Use(middlewares ...Middleware)
+	// RegTask 注册任务
+	RegTask(pattern 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()
+}
+
+// NewExecutor 创建执行器
+func NewExecutor(opts ...Option) Executor {
+	return newExecutor(opts...)
+}
+
+func newExecutor(opts ...Option) *executor {
+	options := newOptions(opts...)
+	e := &executor{
+		opts: options,
+	}
+	return e
+}
+
+type executor struct {
+	opts    Options
+	address string
+	regList *taskList //注册任务列表
+	runList *taskList //正在执行任务列表
+	mu      sync.RWMutex
+	log     Logger
+
+	logHandler  LogHandler   //日志查询handler
+	middlewares []Middleware //中间件
+}
+
+func (e *executor) Init(opts ...Option) {
+	for _, o := range opts {
+		o(&e.opts)
+	}
+	e.log = e.opts.l
+	e.regList = &taskList{
+		data: make(map[string]*Task),
+	}
+	e.runList = &taskList{
+		data: make(map[string]*Task),
+	}
+	e.address = e.opts.ExecutorIp + ":" + e.opts.ExecutorPort
+	go e.registry()
+}
+
+// LogHandler 日志handler
+func (e *executor) LogHandler(handler LogHandler) {
+	e.logHandler = handler
+}
+
+func (e *executor) Use(middlewares ...Middleware) {
+	e.middlewares = middlewares
+}
+
+func (e *executor) Run() (err error) {
+	// 创建路由器
+	mux := http.NewServeMux()
+	// 设置路由规则
+	mux.HandleFunc("/run", e.runTask)
+	mux.HandleFunc("/kill", e.killTask)
+	mux.HandleFunc("/log", e.taskLog)
+	mux.HandleFunc("/beat", e.beat)
+	mux.HandleFunc("/idleBeat", e.idleBeat)
+	// 创建服务器
+	server := &http.Server{
+		Addr:         ":" + e.opts.ExecutorPort,
+		WriteTimeout: time.Second * 3,
+		Handler:      mux,
+	}
+	// 监听端口并提供服务
+	e.log.Info("Starting server at " + e.address)
+	go server.ListenAndServe()
+	quit := make(chan os.Signal)
+	signal.Notify(quit, syscall.SIGKILL, syscall.SIGQUIT, syscall.SIGINT, syscall.SIGTERM)
+	<-quit
+	e.registryRemove()
+	return nil
+}
+
+func (e *executor) Stop() {
+	e.registryRemove()
+}
+
+// RegTask 注册任务
+func (e *executor) RegTask(pattern string, task TaskFunc) {
+	var t = &Task{}
+	t.fn = e.chain(task)
+	e.regList.Set(pattern, t)
+}
+
+// 运行一个任务
+func (e *executor) runTask(writer http.ResponseWriter, request *http.Request) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	req, _ := io.ReadAll(request.Body)
+	param := &RunReq{}
+	err := json.Unmarshal(req, &param)
+	if err != nil {
+		_, _ = writer.Write(returnCall(param, FailureCode, "params err"))
+		e.log.Error("参数解析错误:" + string(req))
+		return
+	}
+	e.log.Info("任务参数:%v", param)
+	if !e.regList.Exists(param.ExecutorHandler) {
+		_, _ = writer.Write(returnCall(param, FailureCode, "Task not registered"))
+		e.log.Error("任务[" + Int64ToStr(param.JobID) + "]没有注册:" + param.ExecutorHandler)
+		return
+	}
+
+	//阻塞策略处理
+	if e.runList.Exists(Int64ToStr(param.JobID)) {
+		if param.ExecutorBlockStrategy == coverEarly { //覆盖之前调度
+			oldTask := e.runList.Get(Int64ToStr(param.JobID))
+			if oldTask != nil {
+				oldTask.Cancel()
+				e.runList.Del(Int64ToStr(oldTask.Id))
+			}
+		} else { //单机串行,丢弃后续调度 都进行阻塞
+			_, _ = writer.Write(returnCall(param, FailureCode, "There are tasks running"))
+			e.log.Error("任务[" + Int64ToStr(param.JobID) + "]已经在运行了:" + param.ExecutorHandler)
+			return
+		}
+	}
+
+	cxt := context.Background()
+	task := e.regList.Get(param.ExecutorHandler)
+	if param.ExecutorTimeout > 0 {
+		task.Ext, task.Cancel = context.WithTimeout(cxt, time.Duration(param.ExecutorTimeout)*time.Second)
+	} else {
+		task.Ext, task.Cancel = context.WithCancel(cxt)
+	}
+	task.Id = param.JobID
+	task.Name = param.ExecutorHandler
+	task.Param = param
+	task.log = e.log
+
+	e.runList.Set(Int64ToStr(task.Id), task)
+	go task.Run(func(code int64, msg string) {
+		e.callback(task, code, msg)
+	})
+	e.log.Info("任务[" + Int64ToStr(param.JobID) + "]开始执行:" + param.ExecutorHandler)
+	_, _ = writer.Write(returnGeneral())
+}
+
+// 删除一个任务
+func (e *executor) killTask(writer http.ResponseWriter, request *http.Request) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	req, _ := io.ReadAll(request.Body)
+	param := &killReq{}
+	_ = json.Unmarshal(req, &param)
+	if !e.runList.Exists(Int64ToStr(param.JobID)) {
+		_, _ = writer.Write(returnKill(param, FailureCode))
+		e.log.Error("任务[" + Int64ToStr(param.JobID) + "]没有运行")
+		return
+	}
+	task := e.runList.Get(Int64ToStr(param.JobID))
+	task.Cancel()
+	e.runList.Del(Int64ToStr(param.JobID))
+	_, _ = writer.Write(returnGeneral())
+}
+
+// 任务日志
+func (e *executor) taskLog(writer http.ResponseWriter, request *http.Request) {
+	var res *LogRes
+	data, err := io.ReadAll(request.Body)
+	req := &LogReq{}
+	if err != nil {
+		e.log.Error("日志请求失败:" + err.Error())
+		reqErrLogHandler(writer, req, err)
+		return
+	}
+	err = json.Unmarshal(data, &req)
+	if err != nil {
+		e.log.Error("日志请求解析失败:" + err.Error())
+		reqErrLogHandler(writer, req, err)
+		return
+	}
+	e.log.Info("日志请求参数:%+v", req)
+	if e.logHandler != nil {
+		res = e.logHandler(req)
+	} else {
+		res = defaultLogHandler(req)
+	}
+	str, _ := json.Marshal(res)
+	_, _ = writer.Write(str)
+}
+
+// 心跳检测
+func (e *executor) beat(writer http.ResponseWriter, request *http.Request) {
+	e.log.Info("心跳检测")
+	_, _ = writer.Write(returnGeneral())
+}
+
+// 忙碌检测
+func (e *executor) idleBeat(writer http.ResponseWriter, request *http.Request) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	defer request.Body.Close()
+	req, _ := io.ReadAll(request.Body)
+	param := &idleBeatReq{}
+	err := json.Unmarshal(req, &param)
+	if err != nil {
+		_, _ = writer.Write(returnIdleBeat(FailureCode))
+		e.log.Error("参数解析错误:" + string(req))
+		return
+	}
+	if e.runList.Exists(Int64ToStr(param.JobID)) {
+		_, _ = writer.Write(returnIdleBeat(FailureCode))
+		e.log.Error("idleBeat任务[" + Int64ToStr(param.JobID) + "]正在运行")
+		return
+	}
+	e.log.Info("忙碌检测任务参数:%v", param)
+	_, _ = writer.Write(returnGeneral())
+}
+
+// 注册执行器到调度中心
+func (e *executor) registry() {
+	req := &Registry{
+		RegistryGroup: "EXECUTOR",
+		RegistryKey:   e.opts.RegistryKey,
+		RegistryValue: "http://" + e.address,
+	}
+	param, err := json.Marshal(req)
+	if err != nil {
+		log.Fatal("执行器注册信息解析失败:" + err.Error())
+	}
+	if e.keepBeat(param) {
+		e.log.Info("connect registry success")
+	}
+	t := time.NewTicker(time.Second * 20)
+	defer t.Stop()
+	for {
+		<-t.C
+		go e.keepBeat(param)
+	}
+}
+
+func (e *executor) keepBeat(param []byte) bool {
+	ok := false
+	result, err := e.post("/api/registry", string(param))
+	if err != nil {
+		e.log.Error("connect registry fail[1]:" + err.Error())
+		return ok
+	}
+	defer result.Body.Close()
+	body, err := io.ReadAll(result.Body)
+	if err != nil {
+		e.log.Error("connect registry fail[2]:" + err.Error())
+		return ok
+	}
+	res := &res{}
+	_ = json.Unmarshal(body, &res)
+	if res.Code != SuccessCode {
+		e.log.Error("connect registry fail[3]:" + string(body))
+		return ok
+	}
+	return true
+}
+
+// 执行器注册移除
+func (e *executor) registryRemove() {
+	t := time.NewTimer(time.Second * 0) //初始立即执行
+	defer t.Stop()
+	req := &Registry{
+		RegistryGroup: "EXECUTOR",
+		RegistryKey:   e.opts.RegistryKey,
+		RegistryValue: "http://" + e.address,
+	}
+	param, err := json.Marshal(req)
+	if err != nil {
+		e.log.Error("执行器移除失败:" + err.Error())
+		return
+	}
+	res, err := e.post("/api/registryRemove", string(param))
+	if err != nil {
+		e.log.Error("执行器移除失败:" + err.Error())
+		return
+	}
+	defer res.Body.Close()
+	body, _ := io.ReadAll(res.Body)
+	e.log.Info("执行器移除成功:" + string(body))
+}
+
+// 回调任务列表
+func (e *executor) callback(task *Task, code int64, msg string) {
+	e.runList.Del(Int64ToStr(task.Id))
+	res, err := e.post("/api/callback", string(returnCall(task.Param, code, msg)))
+	if err != nil {
+		e.log.Error("callback err : ", err.Error())
+		return
+	}
+	defer res.Body.Close()
+	body, _ := io.ReadAll(res.Body)
+	e.log.Info("任务回调成功:" + string(body))
+}
+
+// post
+func (e *executor) post(action, body string) (resp *http.Response, err error) {
+	request, err := http.NewRequest("POST", e.opts.ServerAddr+action, strings.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	request.Header.Set("Content-Type", "application/json;charset=UTF-8")
+	request.Header.Set("XXL-JOB-ACCESS-TOKEN", e.opts.AccessToken)
+	client := http.Client{
+		Timeout: e.opts.Timeout,
+	}
+	return client.Do(request)
+}
+
+// RunTask 运行任务
+func (e *executor) RunTask(writer http.ResponseWriter, request *http.Request) {
+	e.runTask(writer, request)
+}
+
+// KillTask 删除任务
+func (e *executor) KillTask(writer http.ResponseWriter, request *http.Request) {
+	e.killTask(writer, request)
+}
+
+// TaskLog 任务日志
+func (e *executor) TaskLog(writer http.ResponseWriter, request *http.Request) {
+	e.taskLog(writer, request)
+}
+
+// Beat 心跳检测
+func (e *executor) Beat(writer http.ResponseWriter, request *http.Request) {
+	e.beat(writer, request)
+}
+
+// IdleBeat 忙碌检测
+func (e *executor) IdleBeat(writer http.ResponseWriter, request *http.Request) {
+	e.idleBeat(writer, request)
+}

go.mod

+module gitlab.wanzhuangkj.com/tush/xxl-job-executor-go
+
+go 1.14
+
+require github.com/go-basic/ipv4 v1.0.0

go.sum

+github.com/go-basic/ipv4 v1.0.0 h1:gjyFAa1USC1hhXTkPOwBWDPfMcUaIM+tvo1XzV9EZxs=
+github.com/go-basic/ipv4 v1.0.0/go.mod h1:etLBnaxbidQfuqE6wgZQfs38nEWNmzALkxDZe4xY8Dg=

log.go

+package xxl
+
+import (
+	"fmt"
+	"log"
+)
+
+// LogFunc 应用日志
+type LogFunc func(req LogReq, res *LogRes) []byte
+
+// Logger 系统日志
+type Logger interface {
+	Info(format string, a ...interface{})
+	Error(format string, a ...interface{})
+}
+
+type logger struct {
+}
+
+func (l *logger) Info(format string, a ...interface{}) {
+	fmt.Println(fmt.Sprintf(format, a...))
+}
+
+func (l *logger) Error(format string, a ...interface{}) {
+	log.Println(fmt.Sprintf(format, a...))
+}

log_handler.go

+package xxl
+
+import (
+	"encoding/json"
+	"net/http"
+)
+
+/**
+用来日志查询，显示到xxl-job-admin后台
+*/
+
+type LogHandler func(req *LogReq) *LogRes
+
+//默认返回
+func defaultLogHandler(req *LogReq) *LogRes {
+	return &LogRes{Code: SuccessCode, Msg: "", Content: LogResContent{
+		FromLineNum: req.FromLineNum,
+		ToLineNum:   2,
+		LogContent:  "这是日志默认返回，说明没有设置LogHandler",
+		IsEnd:       true,
+	}}
+}
+
+//请求错误
+func reqErrLogHandler(w http.ResponseWriter, req *LogReq, err error) {
+	res := &LogRes{Code: FailureCode, Msg: err.Error(), Content: LogResContent{
+		FromLineNum: req.FromLineNum,
+		ToLineNum:   0,
+		LogContent:  err.Error(),
+		IsEnd:       true,
+	}}
+	str, _ := json.Marshal(res)
+	_, _ = w.Write(str)
+}

middleware.go

+package xxl
+
+// Middleware 中间件构造函数
+type Middleware func(TaskFunc) TaskFunc
+
+func (e *executor) chain(next TaskFunc) TaskFunc {
+	for i := range e.middlewares {
+		next = e.middlewares[len(e.middlewares)-1-i](next)
+	}
+	return next
+}

optinos.go

+package xxl
+
+import (
+	"github.com/go-basic/ipv4"
+	"time"
+)
+
+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"`       //日志目录
+
+	l Logger //日志处理
+}
+
+func newOptions(opts ...Option) Options {
+	opt := Options{
+		ExecutorIp:   ipv4.LocalIP(),
+		ExecutorPort: DefaultExecutorPort,
+		RegistryKey:  DefaultRegistryKey,
+	}
+
+	for _, o := range opts {
+		o(&opt)
+	}
+
+	if opt.l == nil {
+		opt.l = &logger{}
+	}
+
+	return opt
+}
+
+type Option func(o *Options)
+
+var (
+	DefaultExecutorPort = "9999"
+	DefaultRegistryKey  = "golang-jobs"
+)
+
+// ServerAddr 设置调度中心地址
+func ServerAddr(addr string) Option {
+	return func(o *Options) {
+		o.ServerAddr = addr
+	}
+}
+
+// AccessToken 请求令牌
+func AccessToken(token string) Option {
+	return func(o *Options) {
+		o.AccessToken = token
+	}
+}
+
+// ExecutorIp 设置执行器IP
+func ExecutorIp(ip string) Option {
+	return func(o *Options) {
+		o.ExecutorIp = ip
+	}
+}
+
+// ExecutorPort 设置执行器端口
+func ExecutorPort(port string) Option {
+	return func(o *Options) {
+		o.ExecutorPort = port
+	}
+}
+
+// RegistryKey 设置执行器标识
+func RegistryKey(registryKey string) Option {
+	return func(o *Options) {
+		o.RegistryKey = registryKey
+	}
+}
+
+// SetLogger 设置日志处理器
+func SetLogger(l Logger) Option {
+	return func(o *Options) {
+		o.l = l
+	}
+}

task.go

+package xxl
+
+import (
+	"context"
+	"fmt"
+	"runtime/debug"
+)
+
+// TaskFunc 任务执行函数
+type TaskFunc func(cxt context.Context, param *RunReq) string
+
+// Task 任务
+type Task struct {
+	Id        int64
+	Name      string
+	Ext       context.Context
+	Param     *RunReq
+	fn        TaskFunc
+	Cancel    context.CancelFunc
+	StartTime int64
+	EndTime   int64
+	//日志
+	log Logger
+}
+
+// Run 运行任务
+func (t *Task) Run(callback func(code int64, msg string)) {
+	defer func(cancel func()) {
+		if err := recover(); err != nil {
+			t.log.Info(t.Info()+" panic: %v", err)
+			debug.PrintStack() //堆栈跟踪
+			callback(FailureCode, fmt.Sprintf("task panic:%v", err))
+			cancel()
+		}
+	}(t.Cancel)
+	msg := t.fn(t.Ext, t.Param)
+	callback(SuccessCode, msg)
+	return
+}
+
+// Info 任务信息
+func (t *Task) Info() string {
+	return fmt.Sprintf("任务ID[%d]任务名称[%s]参数:%s", t.Id, t.Name, t.Param.ExecutorParams)
+}

task_list.go

+package xxl
+
+import "sync"
+
+//任务列表 [JobID]执行函数,并行执行时[+LogID]
+type taskList struct {
+	mu   sync.RWMutex
+	data map[string]*Task
+}
+
+// Set 设置数据
+func (t *taskList) Set(key string, val *Task) {
+	t.mu.Lock()
+	t.data[key] = val
+	t.mu.Unlock()
+}
+
+// Get 获取数据
+func (t *taskList) Get(key string) *Task {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	return t.data[key]
+}
+
+// GetAll 获取数据
+func (t *taskList) GetAll() map[string]*Task {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	return t.data
+}
+
+// Del 设置数据
+func (t *taskList) Del(key string) {
+	t.mu.Lock()
+	delete(t.data, key)
+	t.mu.Unlock()
+}
+
+// Len 长度
+func (t *taskList) Len() int {
+	return len(t.data)
+}
+
+// Exists Key是否存在
+func (t *taskList) Exists(key string) bool {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	_, ok := t.data[key]
+	return ok
+}

util.go

+package xxl
+
+import (
+	"encoding/json"
+	"strconv"
+)
+
+// Int64ToStr int64 to str
+func Int64ToStr(i int64) string {
+	return strconv.FormatInt(i, 10)
+}
+
+//执行任务回调
+func returnCall(req *RunReq, code int64, msg string) []byte {
+	data := call{
+		&callElement{
+			LogID:      req.LogID,
+			LogDateTim: req.LogDateTime,
+			ExecuteResult: &ExecuteResult{
+				Code: code,
+				Msg:  msg,
+			},
+			HandleCode: int(code),
+			HandleMsg:  msg,
+		},
+	}
+	str, _ := json.Marshal(data)
+	return str
+}
+
+//杀死任务返回
+func returnKill(req *killReq, code int64) []byte {
+	msg := ""
+	if code != SuccessCode {
+		msg = "Task kill err"
+	}
+	data := res{
+		Code: code,
+		Msg:  msg,
+	}
+	str, _ := json.Marshal(data)
+	return str
+}
+
+//忙碌返回
+func returnIdleBeat(code int64) []byte {
+	msg := ""
+	if code != SuccessCode {
+		msg = "Task is busy"
+	}
+	data := res{
+		Code: code,
+		Msg:  msg,
+	}
+	str, _ := json.Marshal(data)
+	return str
+}
+
+//通用返回
+func returnGeneral() []byte {
+	data := &res{
+		Code: SuccessCode,
+		Msg:  "",
+	}
+	str, _ := json.Marshal(data)
+	return str
+}