Yao Job 任务调度框架深度指南：双执行模式、进度追踪与数据库持久化

Agent 框架后端低代码RAG【免费下载链接】yao✨ All your agents and workspaces in one place, on every device you own. Track tasks on a board, accessible from desktop, mobile, browser, or API. Self-hosted.项目地址https://gitcode.com/gh_mirrors/ya/yao点击查看免费下载导读Yao当前仓库 job/ 目录内置了一套完整的任务调度与执行框架 Job Framework它在一个统一的模型体系下解决了任务创建、调度、执行、进度上报、日志记录与状态持久化的全链路问题。本文以 job/README.md 为主体结合 job/types.go、job/worker.go、job/job.go、job/goroutine.go、job/process.go、job/data.go 等源码与 yao/models/job/ 下的数据模型定义带你掌握一次性任务Once、Cron 定时任务与守护任务Daemon的创建方式理解 Goroutine 与独立进程两种执行模式各自的适用场景并学会使用进度追踪、结构化日志与完整的 CRUD API 构建可观测、可恢复的自托管任务系统。一、框架概览一个任务从创建到落库的完整闭环Job Framework 是一套 comprehensive task scheduling and execution framework其设计目标可以概括为四个字一切持久化。从任务的元数据、每次执行的实例记录、执行过程中的进度更新到每一条日志全部写入数据库默认基于__yao.job、__yao.job.category、__yao.job.execution、__yao.job.log四个内置模型因此系统重启后可以借助 job/job.go 中的RestoreJobsFromDatabase()从数据库恢复活跃任务实现定义即持久、恢复即继续。从源码结构看框架按职责划分为五个层次对应 job/README.md 的 Architecture 章节层次文件职责数据层job/data.goJob / Category / Execution / Log 四类实体的全部 CRUD 操作执行层job/execution.go、job/job.go执行实例的创建、任务生命周期Push/Stop/Destroy管理Worker 层job/worker.goWorker 池管理、任务分发与并发控制进度层job/progress.go、job/interfaces.go进度追踪与数据库持久化类型层job/types.go全部数据结构、枚举常量与注册表二、数据模型四个内置模型撑起任务全生命周期数据模型定义位于 yao/models/job/ 目录共四个.mod.yao文件job.mod.yao、category.mod.yao、execution.mod.yao、log.mod.yao。它们与 job/types.go 中的Job、Category、Execution、Log结构体一一对应。2.1 Job任务主体以 yao/models/job/job.mod.yao 为准Job 表承载任务的完整元数据关键字段如下身份与展示job_id64 位唯一字符串标识、name、icon、description、sort、created_by调度配置schedule_type枚举once/cron/daemon默认once、schedule_expressionCron 表达式如0 2 * * *、next_run_at/last_run_at时间戳执行配置mode枚举GOROUTINE/PROCESS默认GOROUTINE、max_worker_nums默认 1、max_retry_count默认 0、default_timeout默认超时秒数可被单次执行覆盖、priority默认 0数值越大优先级越高状态控制status枚举 9 种draft/ready/queued/running/paused/completed/failed/cancelled/disabled默认draft、enabled、system、readonly、current_execution_id扩展字段configJSON 配置、Yao 平台字段__yao_created_by、__yao_updated_by、__yao_team_id、__yao_tenant_id。模型还预置了四个复合索引以优化常见查询路径idx_job_category_status分类状态、idx_job_mode_status模式状态、idx_job_schedule_type_next_run调度类型下次运行时间、idx_job_created_by_status创建人状态并开启了permission与timestamps选项。2.2 Category自动管理的分类体系Categoryjob/types.go用于组织任务。框架实现了按名自动创建的机制在 job/data.go 的GetOrCreateCategory(name, description)中先按name查询不存在则自动创建特殊地名称为Default的分类会固定使用category_id default并标记为system分类见 job/data.go 的ensureCategoryExists。这也解释了 job/job.go 中makeJob的行为当CategoryID与CategoryName均为空时任务默认归入Default分类。2.3 Execution每次执行的实例快照Executionjob/types.go是任务每次运行的实例记录字段极为丰富基本做到一次执行一份审计档案触发信息trigger_category、trigger_source、trigger_context、scheduled_at运行归属worker_id、process_id、parent_execution_id支持父子执行链、retry_attempt时间与结果started_at、ended_at、timeout_seconds、duration毫秒、result、error_info、stack_trace、metrics、context执行配置快照execution_config运行时使用与config_snapshotJSON 落库快照。job/data.go 在查询时会从ConfigSnapshot反序列化还原ExecutionConfig确保即使进程重启执行配置也能完整恢复。2.4 Log结构化执行日志Logjob/types.go记录任务执行过程中的每一条事件字段包括level、message、context、source、execution_id、step、progress、duration、error_code、stack_trace、worker_id、process_id、timestamp、sequence序列号。job/data.go 的ListLogs默认按timestamp倒序分页查询SaveLog则始终追加新记录。三、两种执行模式Goroutine 与独立进程执行模式由ModeType常量定义job/types.goconst ( GOROUTINE ModeType GOROUTINE // Execute using Go goroutine (lightweight, fast) PROCESS ModeType PROCESS // Independent process isolated )两种模式在 job/worker.go 中通过switch w.Mode分流最终分别调用executeInGoroutine与executeInProcess。Worker 池中所有 Worker 默认以GOROUTINE模式启动见 job/worker.go。3.1 Goroutine 模式轻量、快速、共享内存在executeInGoroutinejob/worker.go中框架根据执行配置的类型ExecutionType见 job/types.go分派到三种执行器ExecutionType说明实现processYao 处理器默认类型job/goroutine.goExecuteYaoProcesscommand系统命令job/goroutine.goExecuteSystemCommandfuncGo 函数内存注册表内部使用job/goroutine.goExecuteFuncGoroutine 模式执行 Yao 处理器时通过process.NewWithContext(ctx, ...)创建带上下文的进程对象并利用proc.WithCallback(...)注册回调来接收 Yao 侧上报的进度数据详见第五章。共享数据ExecutionOptions.SharedData会通过proc.WithGlobal/proc.WithSID/proc.WithAuthorized注入进程上下文从而把会话 ID、鉴权信息等一并传递到业务处理器中。3.2 Process 模式独立进程、彻底隔离executeInProcessjob/worker.go会生成proc_xxxxxxxx形式的process_id并调用 job/process.go 中对应的执行器执行 Yao 处理器通过exec.CommandContext(ctx, yao, args...)调用yao run processName args命令工作目录设置为config.Conf.RootYao 应用根目录并通过环境变量YAO_JOB_ID、YAO_EXECUTION_ID以及YAO_JOB_SHARED_key共享数据序列化为 JSON 后注入向子进程传递上下文job/process.go执行系统命令同样以独立子进程方式运行exec.CommandContext并注入相同的环境变量体系job/process.goGo 函数ExecutionTypeFunc在 Process 模式下不受支持源码中会回退到 goroutine 执行器job/worker.go。值得注意的细节是convertArgsForYaoRunjob/process.go基础类型字符串、布尔、整数、浮点直接转为命令行参数而 slice、map、struct 等复杂类型会被 JSON 序列化并以::前缀传入如::{a:1}以便yao run正确解析结构化参数。选型建议短小、高频、对延迟敏感的任务优先选择GOROUTINE模式共享内存、无进程启动开销需要强隔离内存泄漏隔离、崩溃不相互影响或必须借助yao runCLI 能力的任务选择PROCESS模式。四、三种调度类型与创建 API调度类型由ScheduleType常量定义job/types.goonce、cron、daemon。对应三个创建函数job/job.go函数调度类型说明Once(mode ModeType, data map[string]interface{}) (*Job, error)once一次性任务执行完毕即结束Cron(mode ModeType, data map[string]interface{}, expression string) (*Job, error)cron按 Cron 表达式周期执行如0 2 * * *每天凌晨 2 点Daemon(mode ModeType, data map[string]interface{}) (*Job, error)daemon持续运行的守护任务配合ctx.Done()优雅退出源码实现中三个函数本质都是设置mode与schedule_type字段后将 data map 序列化为 JSON 并交给内部makeJob反序列化成Job结构体job/job.go。makeJob还会做一系列默认值填充Status默认draft、MaxWorkerNums默认 1、Priority默认 1、CreatedBy默认system、新任务默认Enabled true。此外还提供了三组创建即保存的快捷函数OnceAndSave、CronAndSave、DaemonAndSavejob/job.go它们创建任务后立即调用SaveJob落库。4.1 快速上手一次性任务// 创建 Goroutine 模式的一次性任务 job, err : job.Once(job.GOROUTINE, map[string]interface{}{ name: Example Job, description: This is an example job, }) handler : func(ctx context.Context, execution *job.Execution) error { execution.Info(Job started) execution.SetProgress(50, In progress...) // 执行业务逻辑 time.Sleep(1 * time.Second) execution.SetProgress(100, Completed) execution.Info(Job completed) return nil } err job.Add(1, handler) // 早期接口形式见下方 API 演进说明 if err ! nil { return err } // 启动任务 err job.Start()API 演进说明README 中的Add(priority, handler)属于早期接口形态。当前仓库源码中job/execution.go 已演进为按执行类型添加执行体的三个方法Add(options *ExecutionOptions, processName string, args ...interface{})执行 Yao 处理器、AddCommand(options, command, args, env)执行系统命令、AddFunc(options, name, fn ExecutionFunc, args)执行 Go 函数其中ExecutionFunc签名是func(ctx *ExecutionContext) errorjob/types.goExecutionContext携带CtxGo 上下文、Execution当前执行实例与Args函数参数。4.2 当前源码形式的完整示例// 一次性任务调用 Yao 处理器 opts : job.NewExecutionOptions().WithPriority(10).AddSharedData(sid, session-001) err j.Add(opts, flows.example.cleanup, map[string]interface{}{days: 30}) // 一次性任务执行系统命令 err j.AddCommand(nil, rm, []string{-rf, /tmp/old}, map[string]string{TZ: Asia/Shanghai}) // 一次性任务执行 Go 函数 err j.AddFunc(nil, myHandler, func(ctx *job.ExecutionContext) error { ctx.Execution.Info(Hello from func, args%v, ctx.Args) return nil }, map[string]interface{}{k: v}) // 推送执行 err j.Push() // 等价于 README 中的 Start()其中ExecutionOptions支持链式调用job/types.goWithPriority(int)设置优先级越高越先执行见 job/job.go 的排序逻辑、WithSharedData(map)/AddSharedData(key, value)注入共享数据。4.3 定时任务与守护任务// Cron 定时任务每天凌晨 2 点执行采用独立进程模式 cronJob, err : job.Cron(job.PROCESS, map[string]interface{}{ name: Cleanup Task, }, 0 2 * * *) err cronJob.Add(nil, flows.cleanup.run) err cronJob.Push()// 守护任务持续运行通过 ctx.Done() 优雅退出 daemonJob, err : job.Daemon(job.GOROUTINE, map[string]interface{}{ name: Monitor Daemon, }) err daemonJob.AddFunc(nil, monitor, func(ctx *job.ExecutionContext) error { ticker : time.NewTicker(5 * time.Second) defer ticker.Stop() for { select { case -ctx.Ctx.Done(): return ctx.Ctx.Err() case -ticker.C: ctx.Execution.Info(Performing monitor check) } } }, nil) err daemonJob.Push()4.4 任务生命周期方法Job 提供的生命周期方法一览Push()job/job.go从数据库加载该任务的全部执行实例按优先级降序排序将任务状态置为ready并落库然后逐一向全局 WorkerManager 提交执行提交失败会汇总错误并清理上下文。多个执行实例从同一个任务上下文context.WithCancel派生保证整体可取消Stop()job/job.go将任务置为disabled取消任务级上下文并逐个取消运行中的执行将queued/running状态改写成cancelled并写入取消日志Destroy()job/job.go先Stop()再置为deleted彻底清理资源链式配置方法SetName、SetDescription、SetCategory、SetMaxWorkerNums、SetMaxRetryCount、SetDefaultTimeout、SetConfig、SetStatus等job/job.go全部返回*Job便于链式调用。五、实时进度追踪三层联动机制进度追踪贯穿业务代码 → 执行器 → 数据库三层job/README.md 中描述的三大特性在源码中均有对应实现实时进度更新业务侧通过execution.SetProgress(progress, message)job/execution.go更新进度该方法同时更新内存中的Execution.Progress、落库保存执行实例并自动写一条Progress: 50% - In progress...级别的日志数据库持久化Progress管理器job/progress.go内部以sync.RWMutex保护进度状态Set()在更新内存值的同时若已设置ExecutionID则加载执行实例并同步保存到数据库Get()供查询当前进度回调支持Goroutine 模式执行 Yao 处理器时proc.WithCallback注册的回调会识别type progress的数据包通过extractProgressDatajob/goroutine.go解析progress与message字段随后更新执行进度、写入日志并落库job/goroutine.go。对于系统命令类执行还提供了外部上报入口UpdateExecutionProgress(executionID, progressData)job/goroutine.go命令脚本可以通过 HTTP API 携带执行 ID 上报进度框架会加载执行实例、解析progress/message并保存。ProgressManager接口job/interfaces.go只定义了一个方法Set(progress int, message string) errorProgress结构体即为其标准实现任务对象通过job.Progress()job/progress.go获取进度管理器实例。六、结构化日志系统七级日志与双通道输出LogLevel定义了完整的七级日志job/types.go按严重程度从高到低为Panic、Fatal、Error、Warn、Info、Debug、Trace。Execution提供一组便捷方法job/execution.goInfo、Debug、Warn、Error、Fatal、Panic、Trace它们最终汇聚到统一的Log(level, format, args...)job/execution.go。该方法的处理流程体现了结构化 双通道的设计将LogLevel映射为字符串级别Fatal/Panic映射为fatalTrace映射为debug构造包含JobID、ExecutionID、WorkerID、ProcessID、当前Progress、Timestamp的Log记录调用SaveLog写入数据库同时输出到系统 logger带上前缀[Job:id][Exec:id]便于在服务端日志中快速检索定位。执行过程中框架还会自动落库关键事件日志执行开始/完成时写入info级别日志含进度与耗时失败时写入error级别日志含 Worker ID 与错误信息见 job/worker.go 的processWork逻辑。七、Worker 管理系统池化、分发与背压7.1 全局单例与默认规模WorkerManager 以全局单例形式存在GetWorkerManagerjob/worker.go并在包初始化时自动Start()。默认 Worker 数量为runtime.NumCPU() * 4job/worker.go兼顾资源利用与系统负载测试场景可通过NewWorkerManagerForTest(maxWorkers)创建非单例实例。7.2 经典 Worker Pool 架构源码实现了 Go 并发编程中最经典的 Worker Pool 模式job/worker.goworkQueue chan *WorkRequest容量为maxWorkers*4的待处理队列允许约 4 倍超额积压workerPool chan chan *WorkRequest空闲 Worker 的注册通道activeWorkers map[string]*Worker活跃 Worker 注册表dispatch()job/worker.go调度协程从workQueue取出请求再从workerPool取一个空闲 Worker 的 JobChannel 投递任务Worker.Start()job/worker.go每个 Worker 循环执行注册自己 → 等待任务 → 处理任务。7.3 提交与背压控制SubmitJobjob/worker.go在提交前检查队列水位queueLen queueCap时直接返回work queue is full (n/cap), please retry later错误实现背压保护避免无界积压随后以非阻塞 goroutine 写入队列若任务上下文已取消则放弃提交。监控接口GetQueueStatus()返回当前队列长度与容量GetActiveWorkers()返回活跃 Worker 数可接入服务健康面板。7.4 执行处理主流程每个任务由processWorkjob/worker.go统一处理流程为置执行状态为running并记录worker_id、started_at→ 更新任务状态为running并记录current_execution_id、last_run_at→ 若配置了DefaultTimeout则用context.WithTimeout包裹执行上下文 → 按模式分发执行 → 计算duration与ended_at→ 成功则置completed并将进度置 100失败则置failed并写入error_info→ 一次性任务完成时任务状态置completed定时/守护任务则回到ready等待下次执行。八、系统自愈能力健康检查与数据清理框架在包初始化时job/job.go启动两个后台守护组件健康检查器job/health.go默认每 5 分钟执行一次健康检查通过NewHealthChecker(interval)创建、StopHealthChecker()/RestartHealthChecker(interval)停止或动态调整间隔job/job.go数据清理器默认保留 90 天执行/日志数据initDataCleaner创建NewDataCleaner(90)并启动测试或运维场景可用ForceCleanup()强制执行清理job/job.go。两者都支持通过测试或动态配置重启为长时间运行的任务系统提供基础的自愈与容量管理能力。九、数据库 CRUD API 速查job/data.go 提供了一整套面向四个模型的分页查询、读写与删除函数README 中列出的核心函数均可在源码中找到对应实现Job 相关ListJobs(param model.QueryParam, page int, pagesize int) (maps.MapStrAny, error)job/data.go分页列出任务并自动批量查询分类名填充category_name字段GetJob(jobID string) (*Job, error)job/data.go按job_id查询并关联分类名SaveJob(job *Job) errorjob/data.go新增ID 0自动生成job_id或更新按job_id定位保护id/created_at不被覆盖若仅填了CategoryName会自动解析为分类 IDRemoveJobs(ids []string) error、GetActiveJobs()、CountJobs(param)。Category 相关GetOrCreateCategory(name, description string) (*Category, error)job/data.go按名查询不存在则自动创建GetCategories(param)、SaveCategory、RemoveCategories、CountCategories。Execution 相关GetExecutions(jobID string) ([]*Execution, error)job/data.go按任务查执行实例并从ConfigSnapshot还原执行配置GetExecution(executionID, param)、SaveExecution、RemoveExecutions、CountExecutions。SaveExecution内部还做了 SQLite 兼容处理将config_snapshot、execution_options、result、error_info等 JSON 字段显式转为字符串存储job/data.go。Log 相关ListLogs(jobID string, param model.QueryParam, page, pagesize)job/data.go按任务分页查日志默认按timestamp倒序SaveLog(log *Log) error始终追加新记录、RemoveLogs(ids []string) error。十、测试覆盖与运行方式框架遵循每个功能模块都有对应测试文件的工程规范job/README.md 的 File Structure 与 Test Coverage 章节实际文件与 job/ 目录一致job/data_test.go job/data_internal_test.goJob / Category / Execution / Log 四类 CRUD 测试job/job_test.go任务管理集成测试job/worker_test.goWorker 生命周期、任务提交、双模式、错误处理、并发测试job/progress_test.go进度管理器、执行中进度、数据库持久化、进度查询测试job/types_test.go类型常量与结构体测试job/health_test.go健康检查器测试。运行测试job/README.md 的 Running Tests 章节# 运行全部 CRUD 测试 go test -v ./job/... -run CRUD # 运行全部类型测试 go test -v ./job/... -run Types|Structure # 运行 Worker 管理测试 go test -v ./job/... -run Worker # 运行进度管理测试 go test -v ./job/... -run Progress # 运行全部测试 go test -v ./job/...环境要求测试依赖数据库模型与平台配置运行前需先加载环境变量job/README.md 的 Environment Requirements 章节source $YAO_ROOT/env.local.sh由于框架持久化依赖__yao.job系列内置模型请确保应用已通过yao migrate完成模型建表相关模型见 yao/models/job/。十一、架构总结与选型要点综合 job/README.md 与源码Job Framework 的 8 项技术特征可归纳为完整 CRUD四个模型全量增删改查均有测试覆盖双执行模式Goroutine 轻量快速、独立进程强隔离模式可随任务指定自动分类管理按名 Get-or-Create默认分类开箱即用实时进度追踪业务代码、Yao 回调、外部 API 三条通道均可更新进度结构化日志七级日志、执行上下文完备、数据库 系统双通道输出并发安全Worker Pool 队列背压 互斥锁保护支持多 Worker 并发执行数据持久化任务、执行实例、进度、日志全部落库重启可恢复完备测试每个模块均有对应单元/集成测试文件。实战选型建议高频轻量任务心跳监控、缓存刷新→GOROUTINEOnce/Daemon强隔离或需 CLI 能力数据清理、批处理脚本→PROCESSCron需要跨执行传递用户会话时使用ExecutionOptions.WithSharedData注入sid、authorized等上下文Goroutine 模式会还原为进程的 Global/SID/Authorized 环境见 job/goroutine.go需要外部脚本上报进度时通过UpdateExecutionProgress提供执行 ID 即可接入高并发场景留意SubmitJob的队列满错误可结合GetQueueStatus()监控积压水位并采取退避重试策略。赞分享Agent 框架后端低代码RAG【免费下载链接】yao✨ All your agents and workspaces in one place, on every device you own. Track tasks on a board, accessible from desktop, mobile, browser, or API. Self-hosted.项目地址https://gitcode.com/gh_mirrors/ya/yao点击查看免费下载相关推荐Delta 金手指模拟器 3 种用法预设库、手输代码与不生效排查Delta 金手指模拟器 3 种用法预设库、手输代码与不生效排查 Delta 是一款跑在 iOS 上的多平台经典游戏模拟器它的金手指Cheat Code游戏开发PowerJob vs XXL-Job新一代分布式任务调度框架深度测评PowerJob vs XXL Job新一代分布式任务调度框架深度测评 在分布式系统架构中任务调度框架扮演着关键角色负责协调各类定时任务、批量处理和分布式任务调度后端Celery 任务执行追踪机制深度解析celery.app.trace 模块源码级指南Celery 任务执行追踪机制深度解析celery.app.trace 模块源码级指南 导读 本文聚焦 Celery 分布式任务队列中承载任务执行追踪tas任务调度后端消息队列上一篇Zotero终极免费学术文献管理工具完整指南下一篇使用NVim-Autopairs让代码对齐变得自动化创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考