Microservices Engine:Executor management

Run the following command to pull the Go version of the XXL-JOB SDK using the latest tag.

go get github.com/xxl-job/xxl-job-executor-go@{latest_tag}

Write the business code.

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"),       // Request address. Obtain it from the connection configuration in the Application Management section of the console.
        xxl.AccessToken("xxxxxxx"),     // Request token. Obtain it from the connection configuration in the Application Management section of the console.
        xxl.ExecutorPort("9999"),       // Default is 9999. This parameter is optional.
        xxl.RegistryKey("golang-jobs"), // Executor name.
        xxl.SetLogger(&logger{}),       // Custom logger.
    )
    exec.Init()
    exec.Use(customMiddleware)
    // Set the handler for viewing logs.
    exec.LogHandler(customLogHandle)
    // Register the task handler.
    exec.RegTask("task.test", task.Test)
    exec.RegTask("task.test2", task.Test2)
    exec.RegTask("task.panic", task.Panic)
    log.Fatal(exec.Run())
}

// Custom log handler.
func customLogHandle(req *xxl.LogReq) *xxl.LogRes {
    return &xxl.LogRes{Code: xxl.SuccessCode, Msg: "", Content: xxl.LogResContent{
        FromLineNum: req.FromLineNum,
        ToLineNum:   2,
        LogContent:  "This is a custom log handler",
        IsEnd:       true,
    }}
}

// Implementation of the xxl.Logger interface.
type logger struct{}

func (l *logger) Info(format string, a ...interface{}) {
    fmt.Println(fmt.Sprintf("Custom log - "+format, a...))
}

func (l *logger) Error(format string, a ...interface{}) {
    log.Println(fmt.Sprintf("Custom log - "+format, a...))
}

// Custom middleware.
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
    }
}

Pull the dependencies.

pip install pyxxl

# To write logs to Redis
pip install "pyxxl[redis]"

# To load configurations from a .env file
pip install "pyxxl[dotenv]"

# Install all features
pip install "pyxxl[all]"

Write the business code.

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",  # If xxl-admin can directly connect to the executor's IP address, you do not need to specify 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 "Success..."

@app.register(name="sync_func")
def test_task4():
    # To view execution logs in xxl-admin, you must use g.logger for log printing. By default, only logs of the INFO level and higher are printed.
    n = 1
    g.logger.info("Job %s get executor params: %s" % (g.xxl_run_data.jobId, g.xxl_run_data.executorParams))
    # If a sync task contains a loop, you must check g.cancel_event in each iteration to support the cancel operation.
    while n <= 10 and not g.cancel_event.is_set():
        # If you do not need to view logs from xxl-admin, you can use your own 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 "Success3"


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