：執行器管理

執行以下命令，使用最新的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())
}

// 自訂Tlog器
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()