@eddyq/nestjs

中文文档

0.2.0发布于 2026/11/05MITTypeScriptCommonJS零依赖构建来源已验证近 30 天下载 2073

Eddyq 的 NestJS 模块—— Eddyq 任务队列的装饰者和 DI

文档目录16 节

支持的后端 forRoot
快开始
模块配置
forRoot(options)
forRootAsync(options) 用于 DI 源配置
仅重写 forRoot
多后端 forRoot - Redis上的网约车,Postgres上的付款
处理错误
合作社取消
行政、统计和时间表
时间表
交易顺序
移徙
磁盘板( Wakeboard)
所需资源
许可证

@eddyq/nestjs (英语)

雀巢JS模块埃迪克语name - 一个粗糙的工作队列运行在邮资, (中文(简体) ), 重写,或两者兼有,由原生节点绑定.

pnpm add @eddyq/queue @eddyq/nestjs

npm 有长期存在的错误 optionalDependencies + 锁文件中断包运预建二进制. 使用pnpm或纱线.

支持的后端 `forRoot`

形状	过	编辑返回	最好还是
仅限邮资	`databaseUrl`	`Eddyq`	交易顺序、耐用批次
仅重写	`redis: { url, line? }`	`EddyqRedis`	高通量、快活、低空
多后端	两个+ `queues: [...]` + 键 `defaultProvider`	`EddyqApp`	在一个app中按队列排列

DI 代号( E)@InjectEddyq()返回任意形状 forRoot 配置隐含着. @InjectQueue('foo') 返回 a QueueHandle 不管怎样,那条路是遵循的在多后端情况下透明地转到右后端.

运行仪表板. @eddyq/wakeboard 在您的 Nest app 下挂载 UI (单位:千美元)/wakeboard (默认情况下)与任何后端形状相悖的, 队列、组、时间表、工作检查、暂停/恢复、取消。停下来写入一次性管理端点。见下文"Dashboard (Wakeboard)".

快开始

// app.module.ts
import { Module } from "@nestjs/common";
import { EddyqModule } from "@eddyq/nestjs";
import { EmailProcessor } from "./email.processor";

@Module({
  imports: [
    EddyqModule.forRoot({
      databaseUrl: process.env.DATABASE_URL!,
      workerConcurrency: 20,
      subscribeTo: ["default", "urgent"],
    }),
  ],
  providers: [EmailProcessor],
})
export class AppModule {}

// email.processor.ts
import { Processor, JobHandler, type JobCall } from "@eddyq/nestjs";

@Processor()
export class EmailProcessor {
  @JobHandler("send.email")
  async send({ payload, id, attempt }: JobCall) {
    // throw to retry with exponential backoff.
    // throw new CancelError(...) to fail permanently.
    // throw new RetryError(..., { delayMs }) to retry at a specific delay.
    await sendgrid.send(payload);
  }
}

// some.controller.ts — enqueueing from a request handler
import { Controller, Post, Body } from "@nestjs/common";
import { InjectEddyq, type Eddyq } from "@eddyq/nestjs";

@Controller("notify")
export class NotifyController {
  constructor(@InjectEddyq() private readonly queue: Eddyq) {}

  @Post()
  async fanout(@Body() body: { to: string; subject: string }) {
    const r = await this.queue.enqueue("send.email", body, {
      uniqueKey: `${body.to}:${Date.now()}`,
      priority: 5,
    });
    return { jobId: r.id };
  }
}

这就是整个地表。正常启动巢穴( E)nest start)——开始 onApplicationBootstrap 模块扫描每个提供者 @Processor() + 键 @JobHandler(kind)登记每个工人,并开始运行时间. 关机时会耗尽飞行中的工作(默认30s宽度)并关闭游泳池.

模块配置

`forRoot(options)`

EddyqModule.forRoot({
  databaseUrl: "postgres://…",

  // Forwarded to `Eddyq.connect` — pool sizing + migration line.
  connectOptions: { maxConnections: 20, line: "main" },

  // Worker runtime (ignored if you have no @JobHandler providers).
  workerConcurrency: 20,            // default 10
  subscribeTo: ["default"],         // default ["default"]
  gracefulShutdownMs: 30_000,       // default 30_000

  // Lifecycle knobs.
  autoStart: true,                  // default true — false = register handlers only
  skipMigrationCheck: false,        // default false — match core's deploy-step guard
  runMigrations: false,             // default false — migrations are a deploy step

  // Cron schedules declared in code. Reconciled at boot; see "Cron schedules".
  schedules: [
    { name: "daily-report", cronExpr: "0 0 8 * * *", kind: "report.generate", payload: {} },
  ],

  // Worker-runtime tuning. Defaults are sensible — only tune with a measured reason.
  tuning: {
    completedRetentionSecs: 3600,    // 1h instead of 24h for a high-volume queue
    staleAfterMs:           300_000, // long handlers — bump from 60s
  },
});

`forRootAsync(options)` 用于 DI 源配置

EddyqModule.forRootAsync({
  imports: [ConfigModule],
  inject: [ConfigService],
  useFactory: (cfg: ConfigService) => ({
    databaseUrl: cfg.getOrThrow("DATABASE_URL"),
    workerConcurrency: cfg.get<number>("QUEUE_CONCURRENCY") ?? 10,
  }),
});

仅重写 `forRoot`

EddyqModule.forRoot({
  redis: {
    url: process.env.REDIS_URL!,
    line: "main",           // hash-tag namespace; distinct lines isolate keyspaces
  },
  workerConcurrency: 20,
  subscribeTo: ["webhooks", "fanout"],
});

同样的巢穴地表—— @InjectEddyq() 返回一个 EddyqRedis,处理器和 @InjectQueue 一样的工作。 Redis 后端跳过迁移 (单位:千美元)runMigrations编号 :skipMigrationCheck 和忽略 connectOptions 池大小( 由 Redis 连接管理器管理) .

多后端 `forRoot` - Redis上的网约车,Postgres上的付款

EddyqModule.forRoot({
  databaseUrl: process.env.DATABASE_URL!,
  redis:       { url: process.env.REDIS_URL!, line: "main" },
  queues: [
    { name: "webhooks", provider: "redis" },
    { name: "payments", provider: "postgres" },
  ],
  defaultProvider: "postgres",   // unrouted queue names land here
});

在一个Nest应用中:

@InjectQueue('webhooks') 返回一个对 Redis 进行排队的控点.
@InjectQueue('payments') 返回一个对 Postgres 进行排序的控点.
通过下列方式注册的单一操作员: @JobHandler('process') 服务岗位那种来自 无论是 后端 — 实际获取的后端引出接头. 使用 payload.queue (或任何有效载荷场)至分支如果种类是共享的.
enqueueBatch 是只使用 Postgres —— 调用 redis 路由控件在运行时丢出一个清晰的消息.
enqueueInTx (和 SELECT eddyq_enqueue(...) SQL 路径) 用于 Postgres 路线的队列。重排队叫它程序员错误.

完整参考应用 : examples/nestjs-mixed。。。 .

处理错误

掌柜可以投出三样东西:

掷出	效果
任意 `Error`	任务重试, 以指数倒置直到 `maxAttempts`
`new CancelError(msg)`	马克永久失败——不再重复
`new RetryError(msg, { delayMs: 60_000 })`	在特定延迟时重试( 如荣誉) `Retry-After`(中文(简体) )

import { CancelError, RetryError } from "@eddyq/nestjs";

@JobHandler("webhook.call")
async call({ payload }: JobCall) {
  const r = await fetch(payload.url, { method: "POST" });
  if (r.status === 429) {
    const after = Number(r.headers.get("retry-after")) * 1000;
    throw new RetryError("rate limited", { delayMs: after });
  }
  if (r.status === 400) throw new CancelError("bad request — no retry");
  if (!r.ok) throw new Error(`HTTP ${r.status}`);
}

合作社取消

打开 app.close() eddyq广播 .abort() 给每个飞行操作员. 结构 signal 关闭 JobCall 把它传给任何接受的人 a 一个 AbortSignal数字 :

@JobHandler("download")
async download({ payload, signal }: JobCall) {
  const r = await fetch(payload.url, { signal });
  return await r.json();
}

行政、统计和时间表

生下来的 Eddyq 客户端被输出到全球——在任何地方注入:

import { InjectEddyq, type Eddyq } from "@eddyq/nestjs";

@Injectable()
export class QueueAdmin {
  constructor(@InjectEddyq() private readonly q: Eddyq) {}

  async pauseIntegrations() {
    await this.q.pauseQueue("integrations");
  }

  async dashboard() {
    const [stats, queues, groups] = await Promise.all([
      this.q.getStats(),
      this.q.listNamedQueues(),
      this.q.listGroups(),
    ]);
    return { stats, queues, groups };
  }
}

见 @eddyq/queue 用于完整方法列表: enqueue, (中文(简体) ), cancel, (中文(简体) ), getStats, (中文(简体) ), listJobs, (中文(简体) ), setGroupConcurrency, (中文(简体) ), setGroupRate, (中文(简体) ), setQueueConcurrency, (中文(简体) ), setQueueTimeout, (中文(简体) ), addSchedule, (中文(简体) ), removeSchedule还有更多.

时间表

宣布他们加入 forRoot。清单在启动时进行调节——添加条目为上调, 删除的被删除。 6个战地sec min hour dom month dow) (中文(简体) ).

import { EddyqModule, type ScheduleDeclaration } from "@eddyq/nestjs";

const schedules: ScheduleDeclaration[] = [
  { name: "daily-report", cronExpr: "0 0 8 * * *", kind: "report.generate", payload: { scope: "daily" } },
];

EddyqModule.forRoot({ databaseUrl: process.env.DATABASE_URL!, schedules });

单人主选跳漏的语义,所以N复制品永远不会双发和长时间停电不会重放每一次被错过的滴答.

交易顺序

在 Redis 上使用 Postgres 后置队列的主要原因: 您的域名写入你的工作在解剖学上是坚定的如果承诺失败,没有发票 财务报告和审定财务报表 没有后续工作。如果任务排队, 发票肯定存在 .

eddyq通过一个 SQL 函数 (单位:千美元)eddyq_enqueue),运行于内部自己的DB客户端的交易——Prisma, Drizzle, Knex, pg, 随便你已经使用 :

// Prisma
await prisma.$transaction(async (tx) => {
  const invoice = await tx.invoice.create({ data });
  await tx.$executeRaw`
    SELECT eddyq_enqueue(
      'send.receipt',
      ${JSON.stringify({ invoiceId: invoice.id })}::jsonb
    )
  `;
});

// Drizzle
await db.transaction(async (tx) => {
  const [invoice] = await tx.insert(invoices).values(data).returning();
  await tx.execute(sql`
    SELECT eddyq_enqueue(
      'send.receipt',
      ${JSON.stringify({ invoiceId: invoice.id })}::jsonb
    )
  `);
});

// pg
await client.query("BEGIN");
const { rows } = await client.query("INSERT INTO invoices (...) RETURNING id", [...]);
await client.query(
  `SELECT eddyq_enqueue('send.receipt', $1::jsonb)`,
  [JSON.stringify({ invoiceId: rows[0].id })],
);
await client.query("COMMIT");

正式签名:

eddyq_enqueue(
  p_kind          text,
  p_payload       jsonb,
  p_queue         text        DEFAULT 'default',
  p_priority      smallint    DEFAULT 0,
  p_max_attempts  integer     DEFAULT 3,
  p_scheduled_at  timestamptz DEFAULT now(),
  p_unique_key    text        DEFAULT null,
  p_group_key     text        DEFAULT null,
  p_tags          text[]      DEFAULT '{}',
  p_metadata      jsonb       DEFAULT '{}'
) RETURNS bigint  -- the new job id, or NULL on unique_key conflict

和一个散装变体 eddyq_enqueue_many(jsonb) 接受 JSONB 系列任务对象,返回 { inserted: N, skipped: N }。。。 .

SQL 路径完全反射出 Rust 路径——相同的INSERT,相同的模式规则实现 group_key, 不变 pg_notify。在 eddyq-core 通过两条路进行同样的工作,并坚持相同的行, 让他们不能漂流.

何时使用:

this.queue.enqueue(...) 通过 @InjectEddyq 简单的火和遗忘,没有外部交易。 90%的病例.
SELECT eddyq_enqueue(...) 当工作完成时必须是带有域写的原子.

移徙

移民是一个 部署步骤问题不是跑步时间。默认 start() 如果注册的迁移没有应用,则路径拒绝启动这使得滚动部署无法运行工人以对抗僵硬的计划.

按照建议顺序,采用三种移徙方式:

一发一发的部署任务 (例如,Kubernetes) Job, ECS 一次性任务) 电话 Eddyq.connect(url).then((q) => q.migrate()) 在工人滚出之前.
这个 eddyq CLI: 爱丽丝 eddyq migrate run --database-url $DATABASE_URL。。。 .
runMigrations: true 内 forRoot - 穿着靴子仅使用用于本地 dev 或测试; 它序列化每个复制品的后靴计划迁移.

设定 skipMigrationCheck: true 如果你管理迁移出带和想要让你的后卫安静下来.

磁盘板( Wakeboard)

@eddyq/wakeboard 在您的 Nest app 下挂载一个完整的管理 UI 。这个自动检测您所连接的后端 forRoot 并且使权利 - 统计、排队、组别、时间表、职位检查、, 暂停/ 恢复、取消、设置货币、设定利率限制 .

import { Module } from "@nestjs/common";
import { EddyqModule } from "@eddyq/nestjs";
import { EddyqWakeboardModule } from "@eddyq/wakeboard";

@Module({
  imports: [
    EddyqModule.forRoot({
      databaseUrl: process.env.DATABASE_URL!,
      redis:       { url: process.env.REDIS_URL!, line: "main" },
      queues: [
        { name: "webhooks", provider: "redis" },
        { name: "payments", provider: "postgres" },
      ],
      defaultProvider: "postgres",
    }),
    EddyqWakeboardModule.forRoot({
      mountPath: "/wakeboard",
      auth: { password: process.env.WAKEBOARD_PASSWORD! },
    }),
  ],
})
export class AppModule {}

点击 http://localhost:3000/wakeboard 你正在看现场排队状态. (原始内容存档于2017-09-21). REST API under /wakeboard/api/* 选择 ?provider=postgres|redis 当两个后端配置时, 查询每个端点的参数 —— 把它连接起来任何你想要的监控. 用这个这样可以省去一千行管理脚手架.

所需资源

20号节点(NestJS 12本身需要20.19或22.12)
PostgreSQL 第 14 (如果使用 Postgres 后端)
Redis \ 7 (如果使用 Redis 后端—— Redis 函数需要 7. 0+)
@nestjs/common 财务报告和审定财务报表 @nestjs/core ^ 10, ^ 11, 或 ^ 12( 等声道)
@eddyq/queue 同一个小版本( pier dep)

许可证

麻省理工学院

版本与发布记录

@eddyq/nestjs 共发布 27 个版本，首次发布于 2026/04/26；过去 12 个月发布了 27 个版本。最新版本附带 npm provenance 构建来源证明，可追溯到公开的 CI 构建记录。

版本	发布日期	距今
`0.2.0`latest	2026/11/05	9 天前
`0.1.14`	2026/10/30	15 天前
`0.1.13`	2026/10/29	16 天前
`0.1.12`	2026/10/29	16 天前
`0.1.11`	2026/10/28	17 天前
`0.1.10`	2026/10/24	20 天前

关于 @eddyq/nestjs 的常见问题

@eddyq/nestjs 是什么？

Eddyq 的 NestJS 模块—— Eddyq 任务队列的装饰者和 DI

如何安装 @eddyq/nestjs？

在项目目录运行 npm install @eddyq/nestjs，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：@eddyq/queue（0.2.0）、@nestjs/core（^10 || ^11 || ^12）、@nestjs/common（^10 || ^11 || ^12）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。

@eddyq/nestjs 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

@eddyq/nestjs 支持 ESM 还是 CommonJS？

以 CommonJS 形式发布，可以直接 require()；在 ES Module 项目中也可以用 import 引入（Node.js 会把 module.exports 作为默认导出）。

@eddyq/nestjs 需要什么版本的 Node.js？

package.json 的 engines 字段要求 Node.js >=20。版本不符时 npm 默认只给出警告，开启 engine-strict 后会拒绝安装。

@eddyq/nestjs 可以商用吗？

@eddyq/nestjs 使用 MIT 许可。宽松许可：允许商业使用、修改和再分发，通常只需在分发时保留原版权与许可声明。以上为概括，不构成法律意见，请以许可原文为准。

@eddyq/nestjs 的下载量有多少？

根据 npm 官方下载统计，@eddyq/nestjs 在 2026/10/13 – 2026/11/11 这 30 天内被下载 2,073 次，日均约 69 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

@eddyq/nestjs 还在维护吗？

最新版本 0.2.0 发布于 2026/11/05（9 天前）；至今共发布 27 个版本，其中过去 12 个月发布了 27 个；首次发布于 2026/04/26。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。

npm install @eddyq/nestjs

pnpmpnpm add @eddyq/nestjs
yarnyarn add @eddyq/nestjs
bunbun add @eddyq/nestjs

peer 依赖@eddyq/queue 0.2.0@nestjs/core ^10 || ^11 || ^12@nestjs/common ^10 || ^11 || ^12

npm 7+ 与 pnpm 8+ 会自动安装必需的 peer 依赖，Yarn 需手动安装。

在你的项目目录中运行此命令。

近 30 天下载

2,073

约 2073 次日均 69 次

统计周期 2026/10/13 – 2026/11/11
数据来自 npm 官方下载统计，1 天前更新

最新版本0.2.0发布于 9 天前

开源许可MIT宽松许可

包体积83.1 kB21 个文件 · 解压后

运行依赖0 个另有 3 个 peer 依赖

模块格式CommonJS

TypeScript自带类型

Node.js>=20

发布版本27 个近 12 个月 27 个

首次发布2026/04/266 个月前

元数据更新2026/11/05

每一个软件包背后，
都有分享与创造的人。