Wasp 后台任务Jobs实战指南基于 PgBoss 的延迟、重试与定时任务【免费下载链接】waspThe batteries-included full-stack framework for the AI era. Develop JS/TS web apps (React, Node.js, and Prisma) using declarative code that abstracts away complex full-stack features like auth, background jobs, RPC, email sending, end-to-end type safety, single-command deployment, and more.项目地址: https://gitcode.com/GitHub_Trending/wa/wasp本文以 Wasp 官方文档 Recurring Jobs 为核心系统讲解 Wasp 后台任务Jobs的完整使用方式如何在main.wasp中声明任务、如何编写 worker 函数、如何在 Operation 与 setupFn 中提交任务以及如何配置定时调度、延迟执行、失败重试与任务跟踪。读完后你将能够在自己的 Wasp 应用中把耗时的业务逻辑发邮件、调用外部 API、异步数据处理安全地挪到后台执行并掌握PgBoss执行器的底层原理、数据库结构与常见坑位。为什么需要后台任务在大多数 Web 应用中用户向服务器发请求、服务器返回数据请求处理得越快应用给人的感觉就越流畅。但有些请求服务器需要额外的时间才能完整处理例如发送一封邮件调用外部 API 的慢速 HTTP 请求处理需要秒级甚至分钟级耗时的数据。此时更好的做法是先尽快给用户返回响应把剩余工作放到后台去做。Wasp 通过后台任务Jobs支持这种模式并且开箱即用地提供以下能力任务在服务器重启后依然保留持久化任务失败后可以自动重试任务可以被延迟到未来的某个时间点执行任务可以按 cron 表达式定时重复执行。快速上手声明并运行你的第一个 Job下面以官方文档中的经典示例展开一个打印消息到控制台、并从数据库返回任务列表的 Job。第 1 步在main.wasp中声明 Job在 Wasp 应用的main.wasp文件中添加job声明job mySpecialJob { executor: PgBoss, perform: { fn: import { foo } from src/workers/bar }, entities: [Task], }这里的三个核心字段executor: PgBoss—— 指定任务执行器目前 Wasp 唯一支持PgBossperform.fn—— 指向实现任务逻辑的 worker 函数从src/下的 NodeJS 文件导入entities: [Task]—— 声明该任务内部需要使用哪些实体Wasp 会把它们注入到 worker 函数的context.entities中。第 2 步实现 worker 函数在src/workers/bar.js中实现fooexport const foo async ({ name }, context) { console.log(Hello ${name}!) const tasks await context.entities.Task.findMany({}) return { tasks } }TypeScript 版本Wasp 会为每个 Job 生成同名、首字母大写的泛型类型例如MySpecialJobimport { type MySpecialJob } from wasp/server/jobs import { type Task } from wasp/entities type Input { name: string; } type Output { tasks: Task[]; } export const foo: MySpecialJobInput, Output async ({ name }, context) { console.log(Hello ${name}!) const tasks await context.entities.Task.findMany({}) return { tasks } }worker 函数规范官方文档明确要求必须是async函数函数的返回值就是该 Job 的执行结果函数接收两个参数args提交任务时传入的数据context: { entities }包含你在 Job 声明中列出的实体的上下文对象。类型安全说明MySpecialJob是 Wasp 生成的泛型类型用于正确标注 worker 函数的入参和返回值类型。它接收两个类型参数——Inputargs的类型和Output返回值的类型。详见下文 JavaScript API 一节。第 3 步在 Operation / setupFn 中提交任务Job 定义完成后就可以在你的 OperationsQuery/Action或 setupFn以及其他任意 NodeJS 代码中提交任务import { mySpecialJob } from wasp/server/jobs const submittedJob await mySpecialJob.submit({ name: Johnny }) // 或者延迟到未来执行.delay() 接受秒数、Date 或 ISO 日期字符串 await mySpecialJob .delay(10) .submit({ name: Johnny })做完以上三步你的 Job 就会被PgBoss执行效果等同于直接调用foo({ name: Johnny })。注意传给 Job 的参数不是强制要求的是否传参完全取决于你如何实现 worker 函数。定时任务Recurring Jobs如果你有需要定期重复执行的工作可以在 Job 声明中添加schedulejob mySpecialJob { executor: PgBoss, perform: { fn: import { foo } from src/workers/bar }, schedule: { cron: 0 * * * *, args: {json { job: args } json} // 可选 } }配置了schedule之后你不需要在 JavaScript/TypeScript 中主动调用任何提交方法——可以想象成foo({ job: args })每小时被自动调度并执行一次。API 参考Job 声明的完整字段一个完整的 Job 声明如下job mySpecialJob { executor: PgBoss, perform: { fn: import { foo } from src/workers/bar, executorOptions: { pgBoss: {json { retryLimit: 1 } json} } }, schedule: { cron: */5 * * * *, args: {json { foo: bar } json}, executorOptions: { pgBoss: {json { retryLimit: 0 } json} } }, entities: [Task], }executor: JobExecutor必填任务执行器负责任务的调度、监控与执行。PgBoss是目前唯一的执行器适用于低流量的生产场景且要求你的app.db.system为PostgreSQL。Wasp 之所以选择 pg-boss 作为第一个任务执行器是因为它借助 PostgreSQL及其SKIP LOCKED机制作为存储与同步手段能在不引入额外基础设施和复杂管理的情况下提供大部分任务队列该有的能力。perform: dict必填fn: ExtImport必填执行具体工作的async函数。由于 Wasp 在服务器端执行 Job导入路径必须指向 NodeJS 文件。它接收两个参数args提交时传入的数据和context: { entities }包含已声明实体的上下文对象。executorOptions: dict提交任务时使用的执行器默认选项会被直接透传给执行器。可在调用时用submit()或schedule中的配置覆盖。其中pgBoss: JSON对应 pg-boss 的 send 选项如retryLimit。schedule: dictcron: string必填5 位占位符格式的 cron 表达式字符串。pg-boss 的调度只精确到分钟级这是其设计如此的原因。需要帮助时可以借助在线 cron 表达式校验工具来生成和检查表达式。args: JSON定时触发时传给perform.fn的参数。executorOptions: dict定时提交时使用的执行器选项。perform.executorOptions是默认值schedule.executorOptions可以覆盖/扩展它其中pgBoss: JSON同样对应 pg-boss 的 send 选项。entities: [Entity]在 Job 内部需要使用到的实体列表用法与 Queries/Actions 中的实体声明 一致。深入 PgBoss工作原理与数据库结构数据存储在独立的pgbossschema 中当你把 pg-boss 加入 Wasp 项目后它会自动在数据库中新增一个名为pgboss的 schema内含一些内部跟踪表包括job和schedule。pg-boss 的大多数表都有一个name列对应你在.wasp文件中定义的 Job 标识符。此外这些表还维护参数、状态、返回值、重试信息、开始与过期时间以及其他 pg-boss 所需的元数据。注意使用PgBoss时数据库的初始化由 Wasp 服务器自动完成不需要在 schema 或迁移文件中手动建表。与服务器同进程运行Wasp 在启动 web 服务器应用的同时会启动 pg-boss两者同时在线。这意味着通过 pg-boss 运行的任务与服务器其他逻辑如 Operations共享 CPU因此应避免在 Job 中运行 CPU 密集型任务Wasp 目前不支持对纯 pg-boss 应用做独立的水平扩展也不支持将其作为独立 worker/进程/线程启动。从源码看pg-boss 的生命周期管理位于 pgBoss.ts 模板startPgBoss()通过boss.start()准备目标 PostgreSQL 数据库并开始任务监控若数据库对象不存在会自动创建且保证整个服务器生命周期内只启动一次。通过PG_BOSS_NEW_OPTIONS定制 pg-boss 实例如果需要定制 pg-boss 实例的创建参数可以设置环境变量PG_BOSS_NEW_OPTIONS其值为一个包含 pg-boss 初始化参数的字符串化 JSON 对象。注意设置该变量会覆盖 Wasp 的所有默认值因此必须同时包含数据库连接信息。从模板源码可见Wasp 的默认配置是{ connectionString: config.databaseUrl }即直接复用 Wasp 的数据库连接串仅当PG_BOSS_NEW_OPTIONS存在时才会用JSON.parse解析并整体替换默认配置。例如要设置连接串并调整任务的归档与删除策略# 在 .env 文件中 PG_BOSS_NEW_OPTIONS{connectionString:postgresql://user:passwordserver:5432/database,archiveCompletedAfterSeconds:86400,deleteAfterDays:30,maintenanceIntervalMinutes:5} # 在 shell 中 PG_BOSS_NEW_OPTIONS{connectionString:postgresql://user:passwordserver:5432/database,archiveCompletedAfterSeconds:86400,deleteAfterDays:30,maintenanceIntervalMinutes:5}如果要在 Heroku 上部署还需要额外设置PG_BOSS_NEW_OPTIONS{connectionString:REGULAR_HEROKU_DATABASE_URL,ssl:{rejectUnauthorized:false}}这是因为 pg-boss 使用的pg扩展默认不会通过 SSL 连接而 Heroku 要求 SSL且其证书是自签名的必须显式关闭证书校验。任务数据的保留与清理默认情况下PgBoss在任务完成或失败后保留数据 12 小时之后移入归档表归档数据再保留 7 天后删除。若要改变这一行为可通过PG_BOSS_NEW_OPTIONS配置归档archiveCompletedAfterSeconds/archiveFailedAfterSeconds与删除deleteAfterSeconds/deleteAfterDays等参数例如上面的示例将已完成任务归档时间设为 86400 秒1 天、删除时间设为 30 天。已知问题重命名带调度的 JobWasp 从 Job 声明中推导任务名并在 pg-boss 各表的name列中使用该名称。如果你重命名了一个原本带有schedule的 Jobpg-boss 仍会继续按旧名字调度但此时已没有对应的 handler任务会变成 stale 并最终过期。解决方法是删除pgbossschema 中schedule表里对应的行如果只是从 Job 声明中移除schedule也需要做同样处理。JavaScript API提交、延迟与跟踪导入 Jobimport { mySpecialJob } from wasp/server/jobsTypeScript 中可同时导入生成的泛型类型import { mySpecialJob, type MySpecialJob } from wasp/server/jobsMySpecialJob这类泛型类型以 Job 声明命名为基础生成例如声明名为mySpecialJob则类型名为MySpecialJob位于wasp/server/jobs模块用于类型安全的 worker 函数标注接收Input与Output两个类型参数。submit(jobArgs, executorOptions)向执行器提交一个 Job可选的jobArgs是 worker 函数接收的 JSON 参数可选的executorOptions是执行器相关的提交选项const submittedJob await mySpecialJob.submit({ job: args })delay(startAfter)延迟 worker 函数的调用时间startAfter可以是整数延迟的秒数默认 0字符串ISO 日期字符串表示在该时间点执行Date在该时间点执行。const submittedJob await mySpecialJob .delay(10) .submit({ job: args }, { retryLimit: 2 })任务跟踪Trackingsubmit()返回一个SubmittedJob实例包含以下字段jobId任务在执行器中的 IDjobName你在.wasp文件中使用的任务名executorName任务执行器名称的 Symbol。此外还有执行器命名空间下的专用对象。对于 pg-boss可以通过pgBoss访问details()获取 pg-boss 特有的任务详细信息cancel()尝试取消一个任务resume()尝试恢复一个已取消的任务。源码视角Wasp 如何解析与生成 Jobs声明解析Wasp.AppSpec.JobWasp 编译器Haskell 实现在 AppSpec/Job.hs 中定义了 Job 声明的数据结构与.wasp语法一一对应Job由executor、perform、schedule可选、entities可选组成JobExecutor目前只有PgBoss一个构造器JSON 解析时只接受字符串PgBoss其他值直接报错Perform包含fn外部导入的函数与可选的executorOptionsSchedule包含cron注释中明确标注为 5 字段 cron 表达式如*/5 * * * *、可选的args与executorOptionsExecutorOptions目前只有pgBoss一个 JSON 字段会被直接透传给执行器。校验规则PgBoss 必须搭配 PostgreSQL在 AppSpec/Valid.hs 的validateAppSpec校验链中有一项专门的validateDbIsPostgresIfPgBossUsed校验——只要项目里声明了 Job使用了 PgBoss数据库系统就必须是 PostgreSQL否则编译器会直接报错。这与文档中要求app.db.system为PostgreSQL的说明完全对应。代码生成server/jobs模块Wasp 的生成器 JobGenerator.hs 负责为每个 Job 生成 SDK 代码生成wasp/server/jobs模块的index.ts导出所有 Job 及其类型为每个 Job 生成独立的.ts文件_job.ts模板其中typeName由 Job 名首字母大写得到例如mySpecialJob→MySpecialJob并把实体的 Prisma 数据、schedulecron、args、options与perform选项编译进生成的代码仅在项目存在任意 Job 时才生成core/job.ts与执行器相关文件core/pgBoss/。这意味着你写的.wasp声明是唯一事实来源类型安全与执行细节全部由编译期生成保证。实战参考Kitchen Sink 示例中的完整用法仓库中的 Kitchen Sink 示例是学习 Jobs 的最佳参考相关文件位于 examples/kitchen-sink/src/features/jobs/。Job 声明jobs.wasp.tsjobs.wasp.ts 演示了三种典型声明job(uppercaseTextJob, { executor: PgBoss, entities: [UppercaseTextRequest], }), job(mySpecialJob, { executor: PgBoss, performExecutorOptions: { pgBoss: { retryLimit: 1 } }, entities: [Task], }), job(mySpecialScheduledJob, { executor: PgBoss, schedule: { cron: 0 * * * *, args: { foo: bar }, executorOptions: { pgBoss: { retryLimit: 2 } }, }, }),可以看到第一个是普通任务第二个通过performExecutorOptions设置了失败重试 1 次第三个配置了每小时执行一次的定时调度并单独为调度设置了重试 2 次。在 Action 中提交任务uppercaseText.ts 演示了Action 立即响应、Job 异步处理的完整模式ActionrequestUppercaseText先把请求写入数据库状态PENDING然后jobs.uppercaseTextJob.submit({ requestId: request.id })提交后台任务立即返回requestId给前端worker 函数uppercaseTextJob随后在后台读取请求、模拟 2 秒处理、把文本转大写后写回数据库状态SUCCESS出错时捕获异常并更新状态为ERROR后重新抛出。这是一个非常适合移植到你自己应用中的异步任务 状态机模板。Worker 函数bar.js 展示了 worker 函数的另一种写法mySpecialJob与mySpecialScheduledJob复用一个内部runBarJob异步函数打印参数与上下文、模拟 4 秒耗时并返回{ hello: world }。小结Wasp 的 Jobs 体系把后台任务从繁琐的队列基础设施中抽象出来你只需在.wasp中做一次声明就能获得持久化、重试、延迟、定时调度与类型安全底层由 PostgreSQL 之上的 PgBoss 提供可靠的存储与同步。使用时请记住三条关键约束数据库必须是 PostgreSQL任务与服务器共享进程与 CPU不要放 CPU 密集型工作重命名带调度的 Job 后需要手动清理pgboss.schedule表。在此基础上Jobs 完全可以承担 Web 应用常见的低流量后台队列场景邮件、外部 API 调用、异步数据转换等。【免费下载链接】waspThe batteries-included full-stack framework for the AI era. Develop JS/TS web apps (React, Node.js, and Prisma) using declarative code that abstracts away complex full-stack features like auth, background jobs, RPC, email sending, end-to-end type safety, single-command deployment, and more.项目地址: https://gitcode.com/GitHub_Trending/wa/wasp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考