使用 Python 执行 XXL-JOB 定时任务（实战二）

青萍叙事2025-07-09

📝 前言

在之前两期我们分别介绍了 XXL-JOB 的基本概念：《定时任务之 XXL-JOB 入门介绍》以及如何通过 Docker 快速部署 XXL-JOB 调度中心服务：《使用 Docker 快速部署 XXL-JOB（实战一）》，这一期我们继续来介绍如何通过 Python 执行 XXL-JOB 的定时任务。

🧭 背景介绍

XXL-JOB 是一款轻量级的分布式任务调度平台，广泛用于企业级 Java 服务定时任务管理。

但默认只支持 Java 执行器，对于希望使用 Python 完成实际任务逻辑的团队来说，并不够友好。

为了填补这一空白，我们可以通过社区实现的 Python 执行器框架 pyxxl，让 Python 脚本也能像 Java 一样被 XXL-JOB 调度和监控。

🔍 使用场景

定期抓取第三方接口数据、写入数据库
定期发送邮件或第三方通知
管理和自动化数据清洗、训练模型等数据工作流

🧩 功能介绍

在使用 Python 对接 XXL-JOB 之前，我们还需要再来理解下它的两个核心组件：执行器（Executor） 和 任务（Job Handler）。

📦 执行器（Executor）

执行器是实际执行任务逻辑的服务节点，它运行在你的服务器或容器中，接收调度中心的指令并调用本地代码。执行器可由 Java、Python、Go 等语言实现。

你可以理解为：”调度中心告诉执行器说‘去执行任务 A’，然后执行器自己去跑任务。”

每个执行器通过名称注册到 XXL-JOB 平台，可以水平扩展多个实例，支持任务路由和负载均衡。

你可以用 Python 编写任务逻辑，只要包装成符合协议的执行器。

PS：这里我们可以借助 pyxxl 框架快速注册执行器。

⚙️ 任务（Job Handler）

任务是你希望调度的具体代码逻辑，在执行器中注册并命名。

一个执行器可以注册多个任务处理函数（Job Handler），每个任务都在调度中心以其名称配置。

任务必须在执行器中通过唯一的名称注册
调度中心只通过任务名称调用，不关心任务实现细节
调度中心可以配置任务的执行时间、频率、失败重试、路由策略等

🌰 举个例子：

执行器：Python 服务 python-send-notify-executor，部署在 192.168.10.2:9999
任务：send_email_job，用于给用户发送邮件通知
调度中心会在每天 09:00 调用这个任务

PS：这一期主要是功能介绍这里导致篇幅有点超了，不过如果能帮助读者更好的理解 XXL-JOB 也是值得的，还请耐心看完~

📋 前提条件

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

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

🚀 详细步骤

1️⃣ 创建执行器（Executor）

我们需要先登录 XXL-JOB 调度中心，手动创建出执行器，然后程序才能注册上来。

打开并登录你的 xxl-job-admin 服务，例如：

1	http://192.168.10.2:8080/xxl-job-admin

点击左侧导航菜单的执行器管理 → 新增，如下图所示：

然后输入执行器的**AppName和名称，注册方式**就保持默认自动注册即可。

2️⃣ 创建任务（Job Handler）

点击左侧导航菜单的任务管理 → 新增，如下图所示：

填写任务参数信息，别看参数很多，只需关注几个核心的即可，如下图：

参数说明：

执行器：选择我们刚上一步创建的执行器。
任务描述：随便写下这个任务主要是做什么的。
负责人：测试环境就只有一个admin可选，生产环境可指定具体的负责人。
调度类型：选择默认CRON，其他还有一个无和固定速度，顾名思义无就是不执行，固定速度就是每隔多少秒就执行一次。
Cron：cron表达式，不知道怎么写的可以点击右边的编辑按钮，直接选择时间。
运行模式：分为BEAN和GLUE两种，BEAN就是javabean，自定义执行函数，GLUE就是将任务代码直接”贴到”调度系统中执行。这里我们选择默认的BEAN即可，在业务代码中实现具体逻辑。
JobHandler：具体要执行的任务函数。
任务参数：下发到执行器任务函数的参数，这里我们可以填写一些需要发送的用户和内容。例如：

{
   "email": "user1@example.com", 
   "name": "张三", 
   "remark": "welcome"
}

3️⃣ 启动任务

到这里还没配置完，我们还需要手动先点击启动任务，因为默认任务创建完成后是STOP停止状态，即使到了时间也不会执行。

点击我们刚创建的任务操作 → 启动即可，如下图：

当状态变为RUNNING即为正常了。

至此，XXL-JOB 调度中心端已经配置完了，接下来该我们来注册执行器并创建自定义任务函数了。

4️⃣ 安装 pyxxl 库

这里我们选择 python 的 pyxxl 框架来注册执行器并执行自定义任务函数。

在执行器服务器（即你打算运行 Python 脚本的机器）上安装 pyxxl：

1	pip install pyxxl

5️⃣ 注册执行器

创建一个main.py文件，添加以下代码配置注册执行器：

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

executor = Executor(
    # 服务器地址，后面需要带上/api/
    xxl_admin_baseurl="http://192.168.10.2:8080/xxl-job-admin/api/",
    # 执行器名称一定要与调度中心的一致
    executor_app_name="python-send-notify-executor",
    # 如果xxl-admin可以直连executor的ip，可以不填executor_listen_host
    # 若应用是在容器内运行的可填写为0.0.0.0，然后添加executor_url=[宿主机IP]:[宿主机端口]
    executor_listen_host="192.168.10.2",
    # 默认监听端口为9999
    executor_listen_port=9999,
    # 若未启用则留空，在部署XXL-JOB时配置的
    access_token="default_token",
)

app = PyxxlRunner(config)

6️⃣ 配置任务函数

继续在mian.py文件下方添加自定义的任务函数，当然你也可以将任务函数单独放到一个文件中再引用。

# 名称需与调度中心任务的JobHandler参数一致
@app.register(name="send_email_job")
async def send_email_task():
    """
    邮件发送任务 - 通过参数传递要发送的邮件信息
    参数格式: {
        "email": "user1@example.com", 
        "name": "张三", 
        "remark": "welcome"
    }
    """
    try:
        params_str = g.xxl_run_data.executorParams
        if not params_str:
            return "未提供邮件发送参数"
            
        params = json.loads(params_str)
        email = params.get("email")
        name = params.get("name", email)
        remark = params.get("remark", "default")
        
        g.logger.info(f"邮件发送任务启动 - 收件人: {email}")
        
        try:
            g.logger.info(f"发送邮件给 {name}({email}) - 备注: {remark}")
            
            # TODO: 发送邮件
            
            g.logger.info(f"邮件发送成功: {email}")
            return f"邮件发送成功, 收件人：{email}, 备注：{remark}"
            
        except Exception as e:
            error_msg = str(e)
            g.logger.error(f"发送给 {email} 失败: {error_msg}")
            return f"邮件发送失败: {error_msg}"
        
    except Exception as e:
        g.logger.error(f"邮件发送任务失败: {e}")
        return f"任务失败: {e}"
 
 
 if __name__ == "__main__":
    app.run_executor()