执行器管理

执行以下命令，使用最新的tag拉取Go版本的XXLJOB SDK。

go get github.com/xxl-job/xxl-job-executor-go@{最新的tag}

编写业务代码。

package main

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

func main() {
    exec := xxl.NewExecutor(
        xxl.ServerAddr("xxxxxx"),       //请求地址，控制台应用管理接入配置获取
        xxl.AccessToken("xxxxxxx"),     //请求令牌，控制台应用管理接入配置获取
        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
    }
}

拉取依赖。

pip install pyxxl

# 如果日志需要写入redis
pip install "pyxxl[redis]"

# 如果需要从.env加载配置
pip install "pyxxl[dotenv]"

# 安装所有功能
pip install "pyxxl[all]"

编写业务代码。

import asyncio
import time

from pyxxl import ExecutorConfig, PyxxlRunner
from pyxxl.ctx import g

config = ExecutorConfig(
    xxl_admin_baseurl="http://xxljob-1b3fd8196eb.schedulerx.mse.aliyuncs.com/api/",
    executor_app_name="xueren-test",
    access_token="default_token",
#    executor_listen_host="0.0.0.0",  # 如果xxl-admin可以直连executor的ip，可以不填写executor_listen_host
)

app = PyxxlRunner(config)


@app.register(name="demoJobHandler")
async def test_task():
    # you can get task params with "g"
    g.logger.info("get executor params: %s" % g.xxl_run_data.executorParams)
    for i in range(10):
        g.logger.warning("test logger %s" % i)
    await asyncio.sleep(5)
    return "成功..."

@app.register(name="sync_func")
def test_task4():
    # 如果要在xxl-admin上看到执行日志，打印日志的时候务必用g.logger来打印，默认只打印info及以上的日志
    n = 1
    g.logger.info("Job %s get executor params: %s" % (g.xxl_run_data.jobId, g.xxl_run_data.executorParams))
    # 如果同步任务里面有循环，为了支持cancel操作，必须每次都判断g.cancel_event.
    while n <= 10 and not g.cancel_event.is_set():
        # 如果不需要从xxl-admin中查看日志，可以用自己的logger
        g.logger.info(
            "log to {} logger test_task4.{},params:{}".format(
                g.xxl_run_data.jobId,
                n,
                g.xxl_run_data.executorParams,
            )
        )
        time.sleep(2)
        n += 1
    return "成功3"


if __name__ == "__main__":
    app.run_executor()