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

type

status

date

slug

summary

📝 前言

在之前两期我们分别介绍了 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 服务，例如：

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

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

2️⃣ 创建任务（Job Handler）

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

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

参数说明：

执行器：选择我们刚上一步创建的执行器。

任务描述：随便写下这个任务主要是做什么的。

负责人：测试环境就只有一个admin可选，生产环境可指定具体的负责人。

调度类型：选择默认CRON，其他还有一个无和固定速度，顾名思义无就是不执行，固定速度就是每隔多少秒就执行一次。

Cron：cron表达式，不知道怎么写的可以点击右边的编辑按钮，直接选择时间。

运行模式：分为BEAN和GLUE两种，BEAN就是javabean，自定义执行函数，GLUE就是将任务代码直接”贴到"调度系统中执行。这里我们选择默认的BEAN即可，在业务代码中实现具体逻辑。

JobHandler：具体要执行的任务函数。

任务参数：下发到执行器任务函数的参数，这里我们可以填写一些需要发送的用户和内容。例如：

3️⃣ 启动任务

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

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

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

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

4️⃣ 安装 pyxxl 库

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

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

5️⃣ 注册执行器

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

6️⃣ 配置任务函数

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

PS：这里为了演示只是打印了日志，并没有实现实际的邮件发送。

7️⃣ 启动程序

执行以下命令，启动我们的 python 程序：

待到达指定的任务执行时间后就会自动执行send_email_task这个函数的代码，可以观察日志查看执行结果。

PS：另外在 XXL-JOB 调度中心上也可以通过左侧菜单的调度日志查看执行情况。

✅ 总结

通过pyxxl，我们可以轻松将 Python 脚本纳入 XXL-JOB 的调度体系，享受可视化管理、失败告警、自动重试等平台能力。

相比 crontab，XXL-JOB 更适合团队协作、集中式运维与任务分布式执行。

下一期我们将继续介绍如何将 XXL-JOB 和自定义的执行器任务部署到 kubernetes 生产环境中，如果感兴趣欢迎关注公众号【青萍叙事】，第一时间收到文章更新通知~

💡

有关文章的任何疑问，欢迎您在底部评论区留言，一起交流~

若文章对您有帮助，欢迎请我喝杯咖啡~