# 分布式任务调度 **Repository Path**: deepeng/neotask ## Basic Information - **Project Name**: 分布式任务调度 - **Description**: NeoTask | Python 轻量级异步任务调度库 · 零依赖 · 支持定时/优先级/重试 · SQLite/Redis 后端 · Celery 轻量替代方案。 - **Primary Language**: Unknown - **License**: MIT - **Default Branch**: main - **Homepage**: https://task.pengline.cn/ - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-05-07 - **Last Updated**: 2026-11-12 ## Categories & Tags **Categories**: Uncategorized **Tags**: 分布式任务调度, 任务排队, 延时任务, 耗时任务, 排队机制 ## README

⚡ NeoTask 分布式任务调度系统

轻量级 Python 异步任务队列管理器 · 无需外部服务 · 开箱即用，支持分布式多节点任务

Python License

PyPI PyPI Downloads GitHub Stars

English • 官网 • 文档 • 社区 • 反馈

> - 🚀 **轻量异步**：纯 Python 实现的异步任务队列，专为 AI 生成、视频处理、数据爬取等高耗时任务设计 > - 📦 **无外部服务依赖**：单节点无需配置 Redis、PostgreSQL 等外部服务，分布式多节点环境需要 Redis 支持 > - ⏰ **全场景调度**：原生支持定时任务、周期任务（Cron-like）与延迟任务，开箱即用、自动重试与异常恢复 > - ⚡ 5 分钟上手：`pip install neotask` 3行代码集成，即可使用内部接口 --- ## 特性 - **无外部服务依赖** - 纯 Python 实现，无需 Redis/PostgreSQL - **即时任务** - 支持优先级调度，高优先级优先执行 - **定时任务** - 支持延时执行、固定间隔、Cron 表达式 - **异步并发** - 基于 asyncio，多 Worker 并发处理 - **自动重试** - 失败任务自动重试，可配置次数 - **持久化** - 内存/SQLite/Redis 多种存储后端 - **分布式** - Redis 共享队列、分布式锁、节点心跳与故障任务回收 - **高性能** - 预取机制、批量操作 - **高可用** - 看门狗、超时检测、任务回收、死信队列 - **事件回调** - 支持任务生命周期事件监听 - **进度上报** - executor 内可发布进度，订阅者收到 `task.progress` 事件 - **自定义失败原因** *(v1.1)* - executor 可通过 `report_failure(reason=...)` 标记业务级失败（如 `qa_failed`），`task.failed` 事件携带原因，与系统异常区分 - **父子关联** *(v1.1)* - 任务携带 `parent_task_id` / `child_count` / `completed_children` 元数据，提供 `get_children()` / `get_parent()` 查询；自动汇总交由调用方或编排层，不在 NeoTask 内实现 ------ ## 应用场景 | 场景 | 说明 | 推荐配置 | 使用入口 | | :--------------------- | :--------------------------- | :---------------------- | :------------- | | **AI 文生图/视频生成** | 耗时任务排队，避免阻塞主流程 | `worker_concurrency=3` | TaskPool | | **批量文件处理** | 转码、压缩、上传等批量操作 | `worker_concurrency=10` | TaskPool | | **网页爬虫调度** | 分布式爬取，防止被封 | `storage_type="redis"` | TaskPool | | **定时报表发送** | 每天9点发送日报 | `cron="0 9 * * *"` | TaskScheduler | | **延迟通知** | 用户操作后5分钟发送提醒 | `delay_seconds=300` | TaskScheduler | | **心跳检测** | 每30秒检测服务健康状态 | `interval_seconds=30` | TaskScheduler | | **后台数据分析** | 夜间执行数据聚合任务 | `cron="0 2 * * *"` | TaskScheduler | --- ## 架构&演进 ```mermaid graph TB subgraph User["用户应用层"] APP[用户代码] end subgraph NeoTask["NeoTask 核心"] TP[TaskPool
即时任务入口] TS[TaskScheduler
定时任务入口] subgraph Core["共享核心组件"] LM[LifecycleManager
任务生命周期管理
含父子关联字段、自定义失败原因] QS[QueueScheduler
优先级 + 延迟队列] WP[WorkerPool
工作池 / 并发控制] FM[FutureManager
异步等待 / 结果回调] end subgraph Planned["v1.1 规划组件"] CK[CheckpointManager
断点续跑] CT[CostTracker
成本与耗时采集] CS[ContextStore
本地上下文存储] end subgraph Internal["内部组件"] EB[EventBus
事件总线] MC[MetricsCollector
内存指标聚合] LF[LockFactory
分布式锁] end EX[TaskExecutor
用户业务逻辑] end subgraph Storage["存储层"] MEM[MemoryStorage] SQLITE[(SQLiteStorage)] REDIS[(RedisStorage)] end APP -->|即时任务| TP APP -->|定时任务| TS TS -->|委托| TP TP --> LM TP --> QS TP --> WP TP --> FM WP --> EX WP --> EB WP --> MC WP --> LF WP -.->|v1.1 规划| CK WP -.->|v1.1 规划| CT WP -.->|v1.1 规划| CS LM --> MEM LM --> SQLITE LM --> REDIS QS --> MEM QS --> SQLITE QS --> REDIS CK -.-> MEM CK -.-> SQLITE CK -.-> REDIS CS -.-> MEM CS -.-> SQLITE CS -.-> REDIS MC -.->|仅内存，不持久化| MEM LF --> REDIS ``` **发展路线图** **图注**：v0.1 – v1.0 为已发布版本；v1.1 – v1.3 为规划中的本地能力、性能与监控迭代；v2.0 的编排集成（Orchestration Integration）为**后移项**，待本地能力稳定且有真实多服务需求时再启动。编排能力（DAG、动态路由、插件系统、Web UI）由上层编排服务负责，不在 NeoTask 范围内。 ```mermaid timeline title NeoTask 架构演进路线图 section v0.1 基础任务池 : 本地内存队列 : 异步执行引擎 : 内存/SQLite存储 section v0.2 可观测性 : 事件总线 : 指标收集 : 健康检查 section v0.3 定时调度 : 延时队列/时间轮 : 周期任务 : Cron表达式 section v0.4 分布式基础 : Redis共享队列 : 分布式锁 section v0.5 性能优化 : 预取机制 : 批量操作 : 连接池优化 section v1.0 高可用保障 : 看门狗续期 : 超时检测 : 故障自动恢复 section v1.1 本地能力补全 : Checkpoint断点续跑 : 成本与耗时采集 : 父子任务关联 : 本地上下文存储 : 自定义失败状态 section v1.2 性能与监控 : 批量写入优化 : 查询索引 : 进度上报节流 : 指标快照接口 : 基准测试套件 section v1.3 导出与可靠性 : Prometheus导出器 : 崩溃恢复边界 : 优雅停机配置 : 重试语义统一 section v2.0 编排集成 : OrchestratorClient : Redis Stream消费者 : 心跳与实例注册 : 远程事件上报 ``` ------ ## 快速上手 详细使用方式 请参阅 [文档](https://task.helpenx.com/docs/quickstart.html) ### 安装 **环境要求**：Python ≥ 3.8。基础包依赖 `aiosqlite`（SQLite 存储）与 `croniter`（Cron 表达式）两个库；Redis、psutil 等为可选扩展，按需安装。 ```sh # 基础安装 pip install neotask # 带 Redis 分布式支持 pip install neotask[redis] # 完整安装 pip install neotask[full] ``` ### 即时任务（TaskPool） ```python from neotask import TaskPool async def process(data): return {"result": "done", "data": data} # 创建任务池 pool = TaskPool(executor=process) # 提交任务 task_id = pool.submit({"id": 123}) # 等待结果 result = pool.wait_for_result(task_id) pool.shutdown() ``` ### 定时任务（TaskScheduler） ```python from neotask import TaskScheduler scheduler = TaskScheduler(executor=process) # 延时 60 秒执行 scheduler.submit_delayed({"id": 123}, delay_seconds=60) # 每 5 分钟执行一次 scheduler.submit_interval({"id": 123}, interval_seconds=300) # 每天 9 点执行 scheduler.submit_cron({"id": 123}, "0 9 * * *") scheduler.shutdown() ``` ### 使用上下文管理器 ```python with TaskPool(executor=process) as pool: task_id = pool.submit({"id": 123}) result = pool.wait_for_result(task_id) ``` ### 使用事件回调 ```python from neotask import TaskPool async def on_task_created(event): print(f"任务创建: {event.task_id}") async def on_task_completed(event): print(f"任务完成: {event.task_id}, 结果: {event.data}") async def on_task_failed(event): print(f"任务失败: {event.task_id}, 错误: {event.data}") pool = TaskPool(executor=my_executor) pool.start() # 注册事件回调 pool.on_created(on_task_created) pool.on_completed(on_task_completed) pool.on_failed(on_task_failed) task_id = pool.submit({"test": "event"}) result = pool.wait_for_result(task_id) ``` ### 任务进度上报 在异步 executor 中调用 `report_progress()`（线程 executor 中用 `report_progress_sync()`） 即可发布进度；通过 `pool.on_progress()` 订阅： ```python from neotask import TaskPool, report_progress async def transcode(data): await report_progress(0.2, "探测中") # ... 执行具体工作 ... await report_progress(0.8, "转码中") return {"ok": True} async def on_progress(event): print(f"{event.task_id}: {event.data['progress']:.0%} {event.data['message']}") pool = TaskPool(executor=transcode) pool.on_progress(on_progress) pool.start() task_id = pool.submit({}) pool.wait_for_result(task_id) ``` 进度会同时写回任务记录（`get_task(task_id)["progress"]`），任务成功时自动收敛为 `1.0`。在任务执行上下文之外调用是安全的空操作（返回 `False`）。 ### 轻量性能自测 官方基准只代表某台机器、某组参数。判断“性能是否符合我的场景”，最快的方法是用自己的 executor 在本机跑一轮： ```bash # 先确认环境：内置 5ms 演示执行器 python -m neotask.benchmark --tasks 200 --concurrency 8 --warmup 10 # 换成真实执行器：模块:函数 或 文件.py:函数（支持 Windows 绝对路径） python -m neotask.benchmark --executor my_tasks:handle --tasks 500 --concurrency 16 # 机器可读输出，便于脚本/CI 横向对比 python -m neotask.benchmark --executor my_tasks:handle --json ``` 也可以用统一 CLI 入口 `neotask --benchmark`（安装后可用，参数与上面完全一致）， 或 `python -m neotask --benchmark`。报告中会列出本次使用的任务数、并发度、预热、 超时、预取、存储、执行器方式与任务数据模板。 代码中也可以直接调用（异步用 `run_benchmark_async`，同步脚本用 `run_benchmark`）： ```python from neotask.benchmark import run_benchmark_async result = await run_benchmark_async(my_executor, tasks=300, concurrency=8) print(result.format()) # result.to_dict() 可直接 json.dumps ``` 口径说明： - 延迟为 `created_at -> completed_at` 的端到端耗时（含排队），只统计成功任务， p50/p95/p99 用于观察长尾。 - 吞吐 =（成功 + 失败）/ 总耗时；失败单独计数，默认 `max_retries=0`，避免重试放大。 - 固定使用 memory 存储，并关闭健康检查/指标/Reporter，只测任务执行主链路。 - 同步执行器会自动放到线程池运行（报告中显示 `thread`），避免阻塞 Worker 事件循环。 - 结果只反映当前机器与当前参数；横向对比请保持 `--tasks` / `--concurrency` / `--payload` 一致，并加 `--warmup` 预热掉首次空闲退避。 ## API 参考 ## API 参考 | 方法 | 说明 | | :----------------------------------------------------------- | :----------------------------------------------------------- | | `pool.submit(data, priority=2, delay=0)` | 提交任务 | | `pool.submit_batch(tasks, priority=2, partial_on_error=False)` | 批量提交任务（默认全有或全无；传 `partial_on_error=True` 可容忍部分失败，详见[批量提交语义](#批量提交语义)） | | `pool.wait_for_result(task_id, timeout=300)` | 等待任务结果 | | `pool.get_status(task_id)` | 获取任务状态 | | `pool.cancel(task_id)` | 取消任务 | | `pool.on_progress(handler)` | 订阅进度事件 | | `scheduler.submit_delayed(data, delay_seconds)` | 提交延时任务 | | `scheduler.submit_interval(data, interval_seconds)` | 提交周期任务 | | `scheduler.submit_cron(data, cron_expr)` | 提交 Cron 任务 | 详细 API 文档见[这里](https://task.helpenx.com/docs/api-reference.html) ### 批量提交语义 `submit_batch` 默认是**全有或全无**语义：任意一条任务在序列化、持久化或入队时失败，整批回滚，并抛出 `BatchSubmitError`，其中包含失败索引与根因。 传入 `partial_on_error=True` 可容忍部分失败： ```python result = pool.submit_batch(tasks, partial_on_error=True) print(result.succeeded) # 实际提交成功的 task_id 列表 print(result.failed) # [{"index": 3, "data": {...}, "error": "queue is full"}] ``` ## 使用须知 以下几处行为容易被误解，特此说明： - **`get_result` 与 `wait_for_result` 的区别**：`wait_for_result(task_id)` 直接返回 executor 的原始结果；`get_result(task_id)` 返回元数据包装字典 `{task_id, status, result, error, created_at, completed_at}`，真实结果在 `result` 字段。 - **事件类型串**：生命周期事件名是带命名空间的 `task.created` / `task.started` / `task.completed` / `task.failed` / `task.cancelled` / `task.retry`（而非裸 `created`）， 建议直接使用 `pool.on_completed(...)` 等便捷方法。 - **优雅关闭的范围**：`shutdown(graceful=True)` 只会把已入队的任务（pending + running） 排空后返回。未到期的延时任务、周期/Cron 调度在关闭期间**不会**被触发——请先 `cancel_periodic` 取消周期调度，或用 `graceful=False` 立即返回。停机时会输出摘要， 明确告知未触发的延时任务数与未入队的 pending 记录数；若希望停机前立即执行未到期 的延时任务，可设置 `drain_delayed_on_shutdown=True`（默认 `False`，保持兼容）。 - **崩溃/重启恢复**：自动恢复只针对已**入队**的任务（其 id 存在于持久化队列中）。 仅通过 `create_pending_task()` 落库、从未调用 `enqueue_task()` 入队的记录不属于队列， 重启后**不会**自动执行——需显式入队。 - **周期任务持久化**：`enable_persistence` **默认开启**。`storage_type="sqlite"` 或 `"redis"` 时，interval / cron 周期任务会持久化并在重启后自动恢复，停机期间错过的槽位按 各任务的 `missed_policy` 处理。`storage_type="memory"` 或显式 `enable_persistence=False` 时无法跨重启；在这种情况下注册周期任务会输出一次明确的 降级告警，而不是重启后静默丢失。 - **周期任务最小间隔**：周期任务由扫描循环驱动，粒度取决于 `SchedulerConfig.scan_interval` （默认 `1.0s`）。`submit_interval` 的间隔小于 `scan_interval` 时可能漏跑；如需更高精度， 请调小 `scan_interval` 或启用时间轮（`enable_time_wheel=True`）。 ## 配置示例 ```python from neotask import TaskPool, TaskPoolConfig config = TaskPoolConfig( worker_concurrency=10, # 并发 Worker 数 max_retries=3, # 重试次数 storage_type="sqlite", # 存储类型 ) pool = TaskPool(executor=process, config=config) ``` 详细使用示例请参阅 [文档](https://task.helpenx.com/docs/configuration.html) ## 贡献指南 ### 开发环境设置 ```python # 克隆仓库 git clone https://github.com/neopen/neotask.git cd neotask # 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 安装开发依赖 pip install -e ".[dev]" # 运行测试 pytest tests/ # 查看测试覆盖率 pytest --cov=src/neotask tests/ # 运行特定模块测试 pytest tests/test_task_pool.py -v pytest tests/test_task_scheduler.py -v ``` ### 项目结构 ``` neotask/ ├── api/ # TaskPool, TaskScheduler ├── core/ # 生命周期、调度器、Future ├── queue/ # 优先级队列、延迟队列 ├── worker/ # Worker 池、监督者、预取 ├── scheduler/ # 定时任务、Cron 解析 ├── executor/ # 异步/线程/进程执行器 ├── storage/ # 内存/SQLite/Redis ├── lock/ # 内存/Redis 分布式锁 ├── event/ # 事件总线 ├── monitor/ # 指标、健康检查、上报 ├── distributed/ # 节点管理、协调器、分片 ├── web/ # 可选 Web UI ├── utils/ # 专用事件循环、Redis/SQLite 代理 ├── models/ # 任务与配置模型 └── common/ # 日志、异常 ``` ### 贡献流程 欢迎提交 Issue 和 Pull Request 1. Fork 项目 2. 创建特性分支 (`git checkout -b feature/amazing`) 3. 提交更改 (`git commit -m 'Add amazing feature'`) 4. 推送分支 (`git push origin feature/amazing`) 5. 提交 Pull Request ### 代码规范 - 遵循 [PEP 8](https://peps.python.org/pep-0008/) 代码风格 - 添加适当的 [类型注解](https://peps.python.org/pep-0484/) - 编写单元测试覆盖新功能（覆盖率 ≥ 80%） - 更新相关文档和示例代码 - 提交信息遵循 [Conventional Commits](https://www.conventionalcommits.org/) ### 测试要求 ```sh # 运行所有测试 pytest tests/ # 运行特定模块测试 pytest tests/test_task_pool.py -v pytest tests/test_task_scheduler.py -v # 运行手动测试 python examples/01_simple.py python examples/05_cron_tasks.py ``` ## 问题反馈 - **提交 Issue**：https://github.com/neopen/neotask/issues - **功能建议**：使用 Enhancement 标签 - **Bug 报告**：使用 Bug 标签并提供复现步骤 - **安全漏洞**：请直接发送邮件至作者邮箱 ------ ## 许可证 MIT License © 2026 neopen ------ ## 致谢 感谢所有贡献者和开源社区的支持。 ------ ## 联系方式 - 项目主页：https://github.com/neopen/neotask - 作者：neopen - 邮箱：helpenx@gmail.com - 文档：https://task.helpenx.com/docs/ - 博客: https://pengline.cn/