AI智能摘要
青萍AI
这里是青萍AI,这篇文章介绍了如何使用Python对接XXL-JOB分布式任务调度平台。作者首先说明XXL-JOB默认仅支持Java执行器,但通过pyxxl框架可实现Python任务调度。文章详细讲解了执行器(Executor)和任务(Job Handler)的概念,并提供了7个具体步骤:从创建执行器、安装pyxxl库到配置任务函数和启动程序。最后指出这种方法适用于数据抓取、邮件通知等定时任务场景,为Python开发者提供了企业级任务调度解决方案。
type
status
date
slug
summary
category
tags
create_time
Jul 8, 2025 12:37 PM
icon
password
my_create_time
📝 前言
在之前两期我们分别介绍了 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 生产环境中,如果感兴趣欢迎关注公众号【青萍叙事】,第一时间收到文章更新通知~
有关文章的任何疑问,欢迎您在底部评论区留言,一起交流~
若文章对您有帮助,欢迎 请我喝杯咖啡~
- 作者:青萍叙事
- 链接:https://blog.lusyoe.com/article/xxl-job-python-executor-guide
- 声明:本文采用 CC BY-NC-SA 4.0 许可协议,转载请注明出处。