:エグゼキュータ管理

次のコマンドを実行して、最新のタグを使用して XXL-JOB Go SDK をプルします。

go get github.com/xxl-job/xxl-job-executor-go@{latest_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)
    // ログ表示用のハンドラを設定します。
    exec.LogHandler(customLogHandle)
    // タスクハンドラを登録します。
    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:  "This is a custom log handler.",
        IsEnd:       true,
    }}
}

// xxl.Logger インターフェイスの実装。
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...))
}

// カスタムミドルウェア。
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 がエグゼキュータの IP アドレスに直接接続できる場合、executor_listen_host を指定する必要はありません。
)

app = PyxxlRunner(config)


@app.register(name="demoJobHandler")
async def test_task():
    # "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():
    # 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))
    # 同期タスクにループが含まれている場合、キャンセル操作をサポートするには、各反復で g.cancel_event をチェックする必要があります。
    while n <= 10 and not g.cancel_event.is_set():
        # xxl-admin からログを表示する必要がない場合は、独自のロガーを使用できます。
        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()