简介
pyxxl 是一个 XXL-JOB 的 Python 执行器实现，可以将 Python 函数注册到 XXL-JOB 调度中心，通过 XXL-JOB-ADMIN 统一管理 Python 定时任务和周期任务。
项目基于 XXL-JOB 提供的 RESTful API 实现，已适配 XXL-JOB 2.2.0 ～ 3.3.2 版本，支持 Python 3.9+。
📦 GitHub 地址：https://github.com/fcfangcc/pyxxl
主要特性
✅ 执行器自动注册到 XXL-JOB-ADMIN
✅ 使用装饰器注册任务（类似 Flask 路由风格）
✅ 任务生命周期管理（启动、取消、完成回调）
✅ 支持全部阻塞策略（串行、丢弃、覆盖）
✅ 异步任务（推荐）和同步任务双模式支持
✅ 在 XXL-JOB 管理界面查看执行日志
✅ 支持 Prometheus 监控指标（需安装 metrics 扩展）
✅ 支持从
.env文件或环境变量读取配置
1. 安装
根据实际需求选择安装方式：
# 基础安装
pip install pyxxl
# 日志写入 Redis
pip install "pyxxl[redis]"
# 从 .env 文件加载配置
pip install "pyxxl[dotenv]"
# 安装全部功能
pip install "pyxxl[all]"
# Prometheus 监控指标
pip install "pyxxl[metrics]"2. 快速开始
基础使用示例
import asyncio
from pyxxl import ExecutorConfig, PyxxlRunner
# 配置执行器
config = ExecutorConfig(
xxl_admin_baseurl="http://localhost:8080/xxl-job-admin/api/",
executor_app_name="xxl-job-executor-sample"
)
app = PyxxlRunner(config)
# 注册异步任务（推荐）
@app.register(name="demoJobHandler")
async def test_task():
await asyncio.sleep(5)
return "执行成功"
# 注册同步任务（适合未全异步化的代码）
@app.register(name="syncJobHandler")
def sync_task():
return "同步任务执行成功"
if __name__ == "__main__":
app.run_executor()3. ExecutorConfig 配置说明
ExecutorConfig 是执行器的核心配置类，支持以下参数：
完整配置示例：
from pyxxl import ExecutorConfig
config = ExecutorConfig(
xxl_admin_baseurl="http://localhost:8080/xxl-job-admin/api/",
executor_app_name="xxl-job-executor-sample",
executor_listen_host="192.168.1.100", # 指定可被 Admin 访问的 IP
executor_listen_port=9999,
debug=True,
)4. 异步任务 vs 同步任务
异步任务（推荐）
如果你的业务代码已经全面异步化，优先使用异步任务，可以充分利用协程的并发优势：
@app.register(name="asyncJobHandler")
async def async_task():
# 调用异步 IO 操作
await some_async_operation()
return "异步执行完成"同步任务
同步任务运行在线程池中，适合无法异步化的传统代码。但需注意：同步任务默认不支持超时和取消信号，如需支持，需要在任务函数中主动检测 g.cancel_event：
import time
from pyxxl import g
@app.register(name="syncLoopHandler")
def sync_loop_task():
task_list = get_pending_tasks()
while not g.cancel_event.is_set() and task_list:
task = task_list.pop()
process(task)
time.sleep(1)
return "处理完成"⚠️ 如果你的同步任务没有实现全异步，直接使用同步函数，否则会阻塞其他任务的执行。
5. 任务参数
XXL-JOB 在调度时可以传递任务参数（executorParam），在 Python 函数中通过 g.job_id 和 g.param 获取：
from pyxxl import g
@app.register(name="paramJobHandler")
async def param_task():
# 获取任务 ID 和参数
job_id = g.job_id
param = g.param # XXL-JOB 界面填写的任务参数
print(f"任务ID: {job_id}, 参数: {param}")
return "OK"6. Prometheus 监控
安装 metrics 扩展后，执行器会自动暴露 Prometheus 格式的指标：
pip install "pyxxl[metrics]"访问地址：
http://<executor_listen_host>:<executor_listen_port>/metrics可通过 Prometheus + Grafana 监控任务执行次数、成功率、耗时等核心指标。
7. 开发与调试
项目提供了一键启动开发环境的脚本（依赖 Docker）：
# 启动 XXL-JOB 调度中心（默认 http://127.0.0.1:8080/xxl-job-admin/）
./init_dev_env.sh
# 安装依赖并启动示例执行器
uv sync --all-extras
python example/executor_app.pyXXL-JOB-Admin 默认账号密码：admin / 123456
兼容性
Python 版本：3.9+（0.3.0 为最后一个支持 Python 3.8 的版本）
XXL-JOB 版本：已测试适配 2.2.0、2.3.0、2.4.0、3.3.2
已知问题
XXL-JOB 界面显示的任务执行时间为回调时间，而非实际开始时间（属于 XXL-JOB 解析层面的问题）
实时日志拉取进度显示异常，刷新页面后可查看完整日志