动态创建 XXL-JOB 定时任务案例（实战四）

青萍叙事2025-07-11

📝 前言

这是定时任务 XXL-JOB 专题的第五篇，回顾一下前几篇内容：

基本概念介绍：《定时任务之 XXL-JOB 入门介绍》
开发/测试环境快速部署：《使用 Docker 快速部署 XXL-JOB（实战一）》
使用Python执行任务：《使用 Python 执行 XXL-JOB 定时任务（实战二）》
在生产环境中使用XXL-JOB：《使用阿里云ACS部署XXL-JOB指南（实战三）》

这一篇过后定时任务专题也暂时告一段落了。

🧭 背景介绍

经过前几期的介绍，我们知道了如果要使用XXL-JOB，需要手动在调度中心控制台创建执行器和任务。

但是在实际工作中，我们可能会遇到需要动态创建任务的场景，比如：

SaaS 平台根据用户设置自动创建定时任务
自动化脚本批量注册、删除任务
配合 GitOps 或 CI/CD 流程动态同步任务配置

这期我们就再来介绍一下如何调用接口动态创建 XXL-JOB 任务（官方并没有提供 OpenAPI）。

📋 前提条件

在开始之前，请确保你已满足以下条件：

已安装部署 XXL-JOB（可参看文章：《使用 Docker 快速部署 XXL-JOB（实战一）》）
已安装 Python 环境（自行查询安装教程）
熟悉基础的 Python 语言

🚀 详细步骤

1️⃣ 安装依赖库

我们这个示例通过 requests 发送接口请求，因此需要先安装依赖库：

1	pip install requests

2️⃣ 获取 XXL-JOB 登录 Token

由于 XXL-JOB 官方并没有提供动态创建任务的 OpenAPI 接口，那我们要如何创建呢？

这里我们有两种选择：

修改 XXL-JOB 源码：使其支持通过OpenAPI来进行创建，成本较大。
模拟登录：通过调用中心的 web 接口来动态创建，毕竟能手动创建的肯定也能自动调(笑)，成本较低。

我们先来模拟登录，示例代码如下：

import requests
import json

# 1. 基本配置
# 后台地址和账号信息
BASE_URL = "http://192.168.10.2:8081/xxl-job-admin"
USERNAME = "admin"
PASSWORD = "123456"

# 创建会话
session = requests.Session()

# 1. 模拟登录
login_url = f"{BASE_URL}/login"
login_data = {
    "userName": USERNAME,
    "password": PASSWORD
}

resp = session.post(login_url, data=login_data)

if "XXL_JOB_LOGIN_IDENTITY" not in session.cookies.get_dict():
    raise Exception("登录失败，请检查账号密码")

print("✅ 登录成功")

3️⃣ 获取执行器 ID

我们在手动创建任务时，可以发现每个任务必须要选择一个执行器，这里是需要传入执行器的ID字段。

我们在登录成功后就可以调用接口，通过执行器的 AppName 查询到ID，示例代码如下：

def get_executor_id(session, appname):
    url = f"{BASE_URL}/jobgroup/pageList"
    # 添加必要的请求头
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'X-Requested-With': 'XMLHttpRequest'
    }
    # 请求参数
    data = {
        "appname": appname, 
        "title": "",
        "start": 0, 
        "length": 10
    }
    resp = session.post(url, data=data, headers=headers)
    data = resp.json()
    for item in data.get("data", []):
        if item["appname"] == appname:
            return item["id"]
    raise Exception("找不到对应执行器")

4️⃣ 构造任务参数并创建任务

在我们拿到执行器ID后，就可以通过session来调用 web 接口创建任务了，示例代码如下：

# 构造添加任务参数
add_job_url = f"{BASE_URL}/jobinfo/add"

payload = {
    "jobGroup": 4,
    "jobDesc": "邮件通知",
    "author": "admin",
    "alarmEmail": "",
    "scheduleType": "CRON",
    "scheduleConf": "0 55 08 12 7 ? 2025",
    "cronGen_display": "0 55 08 12 7 ? 2025",
    "schedule_conf_CRON": "",
    "schedule_conf_FIX_RATE": "",
    "schedule_conf_FIX_DELAY": "",
    "glueType": "BEAN",
    "executorHandler": "send_email_job",
    "executorParam": "",
    "executorRouteStrategy": "FIRST",
    "childJobId": "",
    "misfireStrategy": "FIRE_ONCE_NOW",
    "executorBlockStrategy": "SERIAL_EXECUTION",
    "executorTimeout": 0,
    "executorFailRetryCount": 0,
    "glueRemark": "",
    "glueSource": ""
}

# 3. 发起创建请求
resp = session.post(add_job_url, data=payload)
print("📦 创建任务响应：", resp.text)

主要参数说明：

jobGroup：执行器的ID，可通过动态查询传入
scheduleConf：定时执行时间，示例为一次性任务，在2025-07-12 08:55:00 执行
executorHandler：处理任务的函数名称

5️⃣ 启动任务

最后别忘了，我们手动操作里面还有启动任务这一步，同样我们自动创建的也需要再调一次接口自动启动，示例代码如下：

# 启动任务
start_url = f"{BASE_URL}/jobinfo/start"
start_data = {
    "id": json.loads(resp.text)["content"]
}
resp = session.post(start_url, data=start_data)
print("启动任务响应：", resp.text)