Wasp 后台任务（Jobs）实战指南：基于 PgBoss 的延迟、重试与定时任务

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 唯一支持PgBoss；
• perform.fn—— 指向实现任务逻辑的 worker 函数（从@src/下的 NodeJS 文件导入）；
• entities: [Task]—— 声明该任务内部需要使用哪些实体，Wasp 会把它们注入到 worker 函数的context.entities中。

第 2 步：实现 worker 函数

在src/workers/bar.js中实现foo：

export const foo = async ({ name }, context) => { console.log(`Hello ${name}!`) const tasks = await context.entities.Task.findMany({}) return { tasks } }

TypeScript 版本（Wasp 会为每个 Job 生成同名、首字母大写的泛型类型，例如MySpecialJob）：

import { type MySpecialJob } from 'wasp/server/jobs' import { type Task } from 'wasp/entities' type Input = { name: string; } type Output = { tasks: Task[]; } export const foo: MySpecialJob<Input, 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 函数的入参和返回值类型。它接收两个类型参数——Input（args的类型）和Output（返回值的类型）。详见下文 JavaScript API 一节。

第 3 步：在 Operation / setupFn 中提交任务

Job 定义完成后，就可以在你的 Operations（Query/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 声明中添加schedule：

job 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: dict

• cron: 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:password@server:5432/database","archiveCompletedAfterSeconds":86400,"deleteAfterDays":30,"maintenanceIntervalMinutes":5} # 在 shell 中 PG_BOSS_NEW_OPTIONS='{"connectionString":"postgresql://user:password@server: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 天。

已知问题：重命名带调度的 Job

Wasp 从 Job 声明中推导任务名，并在 pg-boss 各表的name列中使用该名称。如果你重命名了一个原本带有schedule的 Job，pg-boss 仍会继续按旧名字调度，但此时已没有对应的 handler，任务会变成 stale 并最终过期。解决方法是删除pgbossschema 中schedule表里对应的行；如果只是从 Job 声明中移除schedule，也需要做同样处理。

JavaScript API：提交、延迟与跟踪

导入 Job

import { mySpecialJob } from 'wasp/server/jobs'

TypeScript 中可同时导入生成的泛型类型：

import { mySpecialJob, type MySpecialJob } from 'wasp/server/jobs'

MySpecialJob这类泛型类型以 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 })

任务跟踪（Tracking）

submit()返回一个SubmittedJob实例，包含以下字段：

• jobId：任务在执行器中的 ID；
• jobName：你在.wasp文件中使用的任务名；
• executorName：任务执行器名称的 Symbol。

此外还有执行器命名空间下的专用对象。对于 pg-boss，可以通过pgBoss访问：

• details()：获取 pg-boss 特有的任务详细信息；
• cancel()：尝试取消一个任务；
• resume()：尝试恢复一个已取消的任务。

源码视角：Wasp 如何解析与生成 Jobs

声明解析：Wasp.AppSpec.Job

Wasp 编译器（Haskell 实现）在 AppSpec/Job.hs 中定义了 Job 声明的数据结构，与.wasp语法一一对应：

• Job：由executor、perform、schedule（可选）、entities（可选）组成；
• JobExecutor：目前只有PgBoss一个构造器，JSON 解析时只接受字符串"PgBoss"，其他值直接报错；
• Perform：包含fn（外部导入的函数）与可选的executorOptions；
• Schedule：包含cron（注释中明确标注为 5 字段 cron 表达式，如"*/5 * * * *"）、可选的args与executorOptions；
• ExecutorOptions：目前只有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 数据、schedule（cron、args、options）与perform选项编译进生成的代码；
• 仅在项目存在任意 Job 时才生成core/job.ts与执行器相关文件（core/pgBoss/）。

这意味着你写的.wasp声明是"唯一事实来源"，类型安全与执行细节全部由编译期生成保证。

实战参考：Kitchen Sink 示例中的完整用法

仓库中的 Kitchen Sink 示例是学习 Jobs 的最佳参考，相关文件位于 examples/kitchen-sink/src/features/jobs/。

Job 声明（jobs.wasp.ts）

jobs.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