Microservices Engine:Build and connect Java, Go, and Python applications from scratch

Java

1. Add the Maven dependency

Add xxl-job-core to your pom.xml. For the latest version, check the demo project.

<!-- xxl-job-core -->
<dependency>
    <groupId>com.xuxueli</groupId>
    <artifactId>xxl-job-core</artifactId>
    <version>2.2.x</version>
</dependency>

2. Initialize the executor

Create a Spring configuration class that reads connection parameters from your application properties:

@Configuration
public class XxlJobConfig {
    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);

    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;

    @Value("${xxl.job.accessToken}")
    private String accessToken;

    @Value("${xxl.job.executor.appname}")
    private String appname;

    @Value("${xxl.job.executor.address}")
    private String address;

    @Value("${xxl.job.executor.ip}")
    private String ip;

    @Value("${xxl.job.executor.port}")
    private int port;

    @Value("${xxl.job.executor.logpath}")
    private String logPath;

    @Value("${xxl.job.executor.logretentiondays}")
    private int logRetentionDays;

    @Bean
    public XxlJobSpringExecutor xxlJobExecutor() {
        logger.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppname(appname);
        xxlJobSpringExecutor.setAddress(address);
        xxlJobSpringExecutor.setIp(ip);
        xxlJobSpringExecutor.setPort(port);
        xxlJobSpringExecutor.setAccessToken(accessToken);
        xxlJobSpringExecutor.setLogPath(logPath);
        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);

        return xxlJobSpringExecutor;
    }

}

The following table describes each configuration property:

3. Write a task handler

The following example uses the version 2.2.x API. The @XxlJob annotation registers the method as a handler named helloworld, which you reference when creating a task in the console.

XXL-JOB APIs vary by version. Check the demo project for your version.

@Component
public class SampleXxlJob {
    private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);

    @XxlJob("helloworld")
    public ReturnT<String> helloworld(String param) throws Exception {
        XxlJobLogger.log("XXL-JOB, Hello World, start...");
        for (int i = 0; i < 5; i++) {
            XxlJobLogger.log("beat at:" + i);
            TimeUnit.SECONDS.sleep(2);
        }
        System.out.println("XXL-JOB, Hello World, finished");
        return ReturnT.SUCCESS;
    }
}

Go

1. Install the SDK

Pull the Go SDK with the latest tag. For version details, see xxl-job-executor-go.

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

2. Initialize the executor and register task handlers

The following example initializes an executor, registers two task handlers (task.test and task.shardingTest), and sets up a custom log handler and middleware:

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"),       // Admin URL from Connection Configuration in the console
        xxl.AccessToken("xxxxxxx"),     // Access token from Connection Configuration in the console
        xxl.ExecutorPort("9999"),       // Default: 9999 (optional)
        xxl.RegistryKey("golang-jobs"), // Executor name
        xxl.SetLogger(&logger{}),       // Custom logger (optional)
    )
    exec.Init()
    exec.Use(customMiddleware)
    // Set a custom log view handler
    exec.LogHandler(customLogHandle)
    // Register task handlers
    exec.RegTask("task.test", task.Test)
    exec.RegTask("task.shardingTest", task.ShardingTest)
    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,
    }}
}

// Implements 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
    }
}

The following table describes the key executor options:

3. Write task handlers

package task

import (
	"context"
	"fmt"
	xxl "github.com/xxl-job/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"
}

func ShardingTest(cxt context.Context, param *xxl.RunReq) (msg string) {
	fmt.Println("shardingId:" + xxl.Int64ToStr(param.BroadcastIndex) + ", shardingTotal:" + xxl.Int64ToStr(param.BroadcastTotal))
	return "ShardingTest done"
}

Python

1. Install the package

pip install pyxxl

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

# Optional: load configuration from .env files
pip install "pyxxl[dotenv]"

# Install all optional dependencies
pip install "pyxxl[all]"

For version details, see pyxxl.

2. Write task handlers and start the executor

The following example configures the executor, registers two task handlers (one async, one sync), and starts the executor:

import asyncio
import time

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

config = ExecutorConfig(
    xxl_admin_baseurl="http://xxljob-1b3fd81****.schedulerx.mse.aliyuncs.com/api/",
    executor_app_name="your-app-name",
    access_token="default_token",
    # If the admin server can reach the executor directly, leave executor_listen_host empty.
    # executor_listen_host="0.0.0.0",
)

app = PyxxlRunner(config)


@app.register(name="demoJobHandler")
async def test_task():
    # Access task parameters through the global context object 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():
    # Use g.logger for logs visible in the console (INFO level and above)
    n = 1
    g.logger.info("Job %s get executor params: %s" % (g.xxl_run_data.jobId, g.xxl_run_data.executorParams))
    # For sync tasks with loops, check g.cancel_event in each iteration to support cancellation
    while n <= 10 and not g.cancel_event.is_set():
        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 "Success 3"


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

Replace the following placeholders with your actual values: