xxjbob使用指南

采用 XXL-Job 调度中心 + 独立 Go 执行器 + 业务执行的方式：Go 进程常驻，调度中心按配置的周期触发一次过期检查。

阿里云的调度中心是怎么发现服务的

image.png

一、创建执行器

import (
    xxl "github.com/xxl-job/xxl-job-executor-go"
)

// XXLJobAttr XXL-Job 执行器属性
type XXLJobAttr struct {
    RegisterKey string
}

// NewXXLJobExec 创建 XXL-Job 执行器
func NewXXLJobExec(conf *conf.Conf, attr *XXLJobAttr) (xxl.Executor, error) {
    ....
    ....
    exec := xxl.NewExecutor(
        xxl.ServerAddr(AdminUrl),
        xxl.AccessToken(Token),
        xxl.ExecutorPort(RegisterPort),
        xxl.RegistryKey(RegisterKey),
    )

    return exec, nil
}

AdminUrl：是指阿里云xxjob服务的上报地址，也就是我们本地服务上报的地址

image.png

Token：执行器接入该应用时使用的鉴权凭证。Token 用于阿里云平台验证“是否允许接入这个应用”。程序配置的 Token 需要与阿里云该应用的 AccessToken 一致

image.png

RegisterPort：是指你本地服务启动的port,需要将这个port也同时上报给阿里云平台，就知道你服务的port了
RegisterKey：表明“我要注册到哪个应用”，相对应于上图中的AppName

二、执行器初始化

执行器上面创建成功，接着进行初始化，启动注册协程，立即尝试首次上报，此后每 20 秒重复

app.Executor.Init() 

func (e *executor) Init(opts ...Option) {
    ......
    ......
    go e.registry()
}

func (e *executor) registry() {
    .....
    .....
    req := &Registry{
        RegistryGroup: "EXECUTOR",
        RegistryKey:   e.opts.RegistryKey,
        RegistryValue: "http://" + e.address,
    }
    param, err := json.Marshal(req)
    if err != nil {
        log.Fatal("执行器注册信息解析失败:" + err.Error())
    }
    for {
        <-t.C
        t.Reset(time.Second * time.Duration(20)) //20秒心跳防止过期
        func() {
            result, err := e.post("/api/registry", string(param))
            ......
            ......
            e.log.Info("执行器注册成功:" + string(body))
        }()

    }
}

首先是初始化Executor.Init()，可以看到该方法会将三个参数RegistryGroup、RegistryKey、RegistryValue通过post的方式注册到阿里云的xxjjob服务上

RegistryGroup：EXECUTOR，表示执行器，使用默认值即可
RegistryKey：该值为阿里云上创建的应用的appName值，这俩值必须相同，用于本地服务和阿里云xxjob的应用做关联。

RegistryValue：这个值是服务启动后的ip地址**

调用AdminUrl/api/registry地址后，注册中心可以拿到服务上报的ip、port、RegisterKey以及token
注册中心通过RegisterKey匹配应用的appName，获取到对应的应用
根据应用的token和上报的token，是否匹配，匹配则应用关联成功

三、执行器注册任务

上面俩步创建并且初始化了执行器，但是执行器不知道具体执行什么业务呀，所以调用Executor.RegTask就是指定要执行的任务。

type Executor interface {
    // Init 初始化
    Init(...Option)
    // LogHandler 日志查询
    LogHandler(handler LogHandler)
    // Use 使用中间件
    Use(middlewares ...Middleware)
    // RegTask 注册任务
    RegTask(pattern string, task TaskFunc)
    // RunTask 运行任务
    RunTask(writer http.ResponseWriter, request *http.Request)
    // KillTask 杀死任务
    KillTask(writer http.ResponseWriter, request *http.Request)
    // TaskLog 任务日志
    TaskLog(writer http.ResponseWriter, request *http.Request)
    // Beat 心跳检测
    Beat(writer http.ResponseWriter, request *http.Request)
    // IdleBeat 忙碌检测
    IdleBeat(writer http.ResponseWriter, request *http.Request)
    // Run 运行服务
    Run() error
    // Stop 停止服务
}

app.Executor.RegTask(hanlderName, taskFuncion)

hanlderName: 指触发任务时，执行器该执行的哪个方法来执行任务，hanlderName必须和创建任务时的jobhandler保持一致，最终执行时就会执行taskFuncion方法了。

四：运行本地服务

xxjob的方式本质就是运用一个http的服务，所以我们肯定是需要启动一个http服务,调用Executor.Run()即可启动一个常驻进程的http服务了

Executor.Run()

func (e *executor) Run() (err error) {
    // 创建路由器
    mux := http.NewServeMux()
    // 设置路由规则
    mux.HandleFunc("/run", e.runTask)
    mux.HandleFunc("/kill", e.killTask)
    mux.HandleFunc("/log", e.taskLog)
    mux.HandleFunc("/beat", e.beat)
    mux.HandleFunc("/idleBeat", e.idleBeat)
    // 创建服务器
    server := &http.Server{
        Addr:         ":" + e.opts.ExecutorPort,
        WriteTimeout: time.Second * 3,
        Handler:      mux,
    }
    // 监听端口并提供服务
    e.log.Info("Starting server at " + e.address)
    go server.ListenAndServe()
    quit := make(chan os.Signal)
    signal.Notify(quit, syscall.SIGKILL, syscall.SIGQUIT, syscall.SIGINT, syscall.SIGTERM)
    <-quit
    e.registryRemove()
    return nil
}

可以看到run创建了一个http服务，其中服务的port正是我们前边上报执行器的port, 同时注册了一些路由，当阿里云平台执行任务时其实就是调的本地服务的/run路由，接着就会执行这里了app.Executor.RegTask(hanlderName, taskFuncion)

五：xxjJob理解

就是起一个http服务和指定执行任务的路由，阿里云服务触发任务时，就调用这个http服务的路由。我们之前可以更多的是写一个脚本，然后使用cron来定时执行脚本

六、阿里云创建xxjob流程

应用

创建应用：主要用于关联本地服务注册的应用和验证应用的合法性

任务

主要用于关联任务，同时指定任务执行时调用的本地方法，以及控制任务执行的cron和路由策略