# @clipboard-health/mongo-jobs 中文文档

规范地址：https://npmcn.com/package/%40clipboard-health/mongo-jobs/

版本：2\.8\.2
npm 元数据更新时间：2026-10-21T19:47:16.359Z
中文译文更新时间：2026-11-10T01:56:02.556Z
软件包许可：MIT

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40clipboard-health/mongo-jobs
原文对照：https://npmcn.com/package/%40clipboard-health/mongo-jobs/?lang=original
项目主页：https://github.com/ClipboardHealth/mongo-jobs#readme
源码仓库：https://github.com/ClipboardHealth/mongo-jobs

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：156 kB，69 个文件
- 模块格式：CommonJS
- TypeScript 类型：包内自带
- 运行依赖：3 个
- peer 依赖：mongodb \>=6\.9，mongoose \>=8\.7，@opentelemetry/api \>=1\.0
- 首次发布：2025-10-15T12:27:54.873Z
- 最新版本发布：2026-10-21T19:47:16.194Z
- 近 30 天下载量：15145（2026-10-10 至 2026-11-08，npm 官方下载统计）
- 已发布版本数：366（近 12 个月 344 个）

## 中文简介

MongoDB 驱动的背景工作 .

## 安装

```sh
npm install @clipboard-health/mongo-jobs
```

也可以使用：`pnpm add @clipboard-health/mongo-jobs`、`yarn add @clipboard-health/mongo-jobs`、`bun add @clipboard-health/mongo-jobs`

## README

# @clipboard-health/mongo-jobs (英语)

> 一个有TypeScript支持的为Node.js提供强健的、由MongoDB支持的背景任务处理库

## 特征

-   **可靠工作处理** - 在蒙戈DB上建造,用于长期可靠的工作储存
-   **自动重试** - 具有可配置最大尝试的标志性反冲
-   **延迟工作** - 在具体时间安排工作
-   **独一无二的工作** - 确保只有一个工作被排队或运行
-   **编程** - 内置支持带有克龙表达式的经常性工作
-   **职类** - 将工作分成若干组,每个组有专门的工人
-   **交易支持** - 在MongoDB会议上安排工作
-   **类型安全** - 配有强型工作有效载荷的全类型脚本支持
-   **货币管制** - 每组配置工人货币
-   **观察性** - 内设计量报告和伐木支助

## 安装

```bash
npm install @clipboard-health/mongo-jobs
```

## 快速启动

### 1\. 界定工作管理人

工作处理者执行 `HandlerInterface` 并定义您的工作处理方式 :

```ts
import type { HandlerInterface } from "@clipboard-health/mongo-jobs";

export interface WelcomeEmailData {
  userId: string;
  email: string;
}

export class WelcomeEmailJob implements HandlerInterface<WelcomeEmailData> {
  public name = "WelcomeEmailJob";
  public maxAttempts = 3;

  public async perform({ userId, email }: WelcomeEmailData) {
    await this.sendEmail(email, `Welcome, user ${userId}!`);
  }

  private async sendEmail(_to: string, _message: string) {
    // Email sending logic
  }
}
```

### 2\. 创建服务和登记处理员

创建一个 `BackgroundJobs` 实例并登记您的处理器到组:

```ts
import { BackgroundJobs } from "@clipboard-health/mongo-jobs";

import { WelcomeEmailJob } from "./welcomeEmailJob";

const backgroundJobs = new BackgroundJobs();

backgroundJobs.register(WelcomeEmailJob, "emails");

export { backgroundJobs };
```

### 3\. 安排工作

在要处理的队列中添加任务 :

```ts
import { backgroundJobs } from "./jobsRegistry";
import { WelcomeEmailJob } from "./welcomeEmailJob";

await backgroundJobs.enqueue(WelcomeEmailJob, {
  userId: "123",
  email: "user@example.com",
});
```

### 4\. 启动工人

从队列开始处理任务 :

```ts
import { backgroundJobs } from "./jobsRegistry";

await backgroundJobs.start(["emails"], {
  maxConcurrency: 10,
});
```

## 使用量

### 创造就业

工作被定义为: 执行 `HandlerInterface`数字 :

```ts
import type { BackgroundJobType, HandlerInterface } from "@clipboard-health/mongo-jobs";

export interface MyJobData {
  userId: string;
  action: string;
}

export class MyJob implements HandlerInterface<MyJobData> {
  // Required: unique name for this job type
  public name = "MyJob";

  // Optional: max retry attempts (default: 10)
  public maxAttempts = 5;

  // Required: the actual job logic
  public async perform(data: MyJobData, job?: BackgroundJobType<MyJobData>) {
    // Job implementation
    console.log(`Processing ${data.action} for user ${data.userId}`);

    // Optional: access job metadata
    if (job) {
      console.log(`Job ID: ${job._id.toString()}`);
      console.log(`Attempt: ${job.attemptsCount}`);
    }
  }
}
```

#### 任务处理器选项

-   **`name`** (需要):工作类型的唯一标识符
-   **`maxAttempts`** (可选) : 在将任务标记为失败之前, 尝试次数最多 。 默认为10. 使用指数后退: 2^ 尝试秒
-   **`perform`** (需要):执行任务逻辑的 Async 函数
    -   `data`: 工作有效载荷在排队时通过
    -   `job`: 关于任务执行的可选元数据(ID,尝试,时间戳等)

### 注册工作

注册到 `BackgroundJobs` 实例并分配给处理组:

```ts
import { BackgroundJobs } from "@clipboard-health/mongo-jobs";

import { CleanupJob } from "./jobs/cleanupJob";
import { EmailJob } from "./jobs/emailJob";
import { ReportJob } from "./jobs/reportJob";
import { SmsJob } from "./jobs/smsJob";

const backgroundJobs = new BackgroundJobs();

// Register jobs to groups
backgroundJobs.register(EmailJob, "notifications");
backgroundJobs.register(ReportJob, "reports");
backgroundJobs.register(CleanupJob, "maintenance");

// You can register multiple jobs to the same group
backgroundJobs.register(SmsJob, "notifications");
```

团体允许您:

-   共同组织相关工作
-   为特定的工作类型运行专职工人
-   每组控制货币
-   独立调整不同工作类型

#### 有依赖性的工作

如果您的工作需要依赖性(如服务,数据库连接等)通过构建器,您必须注册一个实例而不是类:

```ts
import { BackgroundJobs } from "@clipboard-health/mongo-jobs";

import { EmailServiceJob } from "./jobs/emailServiceJob";

const backgroundJobs = new BackgroundJobs();

// For jobs with constructor dependencies, register an instance
const emailService = {
  async send(to: string, subject: string, body: string) {
    console.log(`Sending email to ${to}: ${subject} : ${body}`);
  },
};

backgroundJobs.register(new EmailServiceJob(emailService), "notifications");
```

有依赖关系的示例工作 :

```ts
import type { HandlerInterface } from "@clipboard-health/mongo-jobs";

interface EmailService {
  send: (to: string, subject: string, body: string) => Promise<void>;
}

export interface EmailServiceJobData {
  to: string;
  subject: string;
  body: string;
}

export class EmailServiceJob implements HandlerInterface<EmailServiceJobData> {
  public name = "EmailServiceJob";
  public maxAttempts = 3;

  public constructor(private readonly emailService: EmailService) {}

  public async perform({ to, subject, body }: EmailServiceJobData) {
    await this.emailService.send(to, subject, body);
  }
}
```

**重要**:在注册工作实例时,库会直接使用实例,而不是即时使用类. 这意味着:

-   在这个过程中,所有职务处决都采用同样的情况
-   所有被处决者都有共同的依赖性
-   你的工作班应该无国籍(所有州应该来自 `data` 参数)

**说明**: 即使在注册实例时,您仍然可以使用类、实例或处理器名称排列工作:

```ts
// All of these work, regardless of whether you registered a class or instance
await backgroundJobs.enqueue(EmailServiceJob, data); // By class (type inferred)
await backgroundJobs.enqueue(emailServiceJobInstance, data); // By instance (type inferred)
await backgroundJobs.enqueue<EmailServiceJobData>("EmailServiceJob", data); // By name (explicit generic required)
```

被排列的类/入门/名只用于查询已注册的处理器. 这个 **登记** 案件总是用于执行,而不是交给 `enqueue()`。 。 。 .

### 排序任务

添加任务到处理的队列中 :

```ts
import { backgroundJobs } from "./jobsRegistry";
import { MyJob } from "./myJob";

// Basic enqueue
await backgroundJobs.enqueue(MyJob, {
  userId: "123",
  action: "process",
});
```

```ts
import type { ClientSession } from "mongodb";

import { backgroundJobs } from "./jobsRegistry";
import { MyJob } from "./myJob";

declare const mongoSession: ClientSession;

// Enqueue with options
await backgroundJobs.enqueue(
  MyJob,
  { userId: "123", action: "process" },
  {
    // Schedule for later
    startAt: new Date("2024-12-31T23:59:59Z"),

    // Ensure uniqueness (see uniqueness section below)
    unique: "user-123-process",

    // Use within a MongoDB transaction
    session: mongoSession,
  },
);
```

```ts
import { backgroundJobs } from "./jobsRegistry";
import type { MyJobData } from "./myJob";

// Enqueue by job name requires explicit generic for type safety
await backgroundJobs.enqueue<MyJobData>("MyJob", { userId: "123", action: "process" });
```

#### 排序选项

-   **`startAt`**: 安排在特定时间运行的任务. 默认即时
-   **`unique`**: 确保只存在一个工作实例(见工作独有性一节)
-   **`session`**: 用于创造交易性工作机会的MongoDB会话

### 创办工人

从一个或多个组开始处理任务 :

```ts
import { backgroundJobs } from "./jobsRegistry";

// Start a worker for specific groups
await backgroundJobs.start(["notifications", "reports"], {
  maxConcurrency: 20,
});
```

```ts
import { backgroundJobs } from "./jobsRegistry";

// Start with all available options
await backgroundJobs.start(["notifications"], {
  // Maximum concurrent jobs (default: 10)
  maxConcurrency: 10,

  // Time to wait when no jobs available, in ms (default: 10000)
  newJobCheckWaitMS: 5000,

  // Use MongoDB change streams for instant job detection (default: true)
  useChangeStream: true,

  // Lock timeout for stuck jobs, in ms (default: 600000 = 10 minutes)
  lockTimeoutMS: 300_000,

  // Interval to check for stuck jobs, in ms (default: 60000 = 1 minute)
  unlockJobsIntervalMS: 30_000,

  // Interval to refresh queue list, in ms (default: 30000 = 30 seconds)
  refreshQueuesIntervalMS: 60_000,

  // Exclude specific queues from processing
  exclude: ["low-priority-queue"],

  // How to pick the next queue to take a job from (default: "random")
  queueSelectionStrategy: "leastInFlight",
});
```

```ts
import { backgroundJobs } from "./jobsRegistry";

// Graceful shutdown
await backgroundJobs.stop(30_000); // Wait up to 30 seconds for jobs to complete
```

#### 工人选项

-   **`maxConcurrency`**: 要同时处理的工作数量
-   **`useChangeStream`**: 启用使用 MongoDB 更改流的即时任务检测 。 何时 `true`在增加新工作时立即通知工人
-   **`newJobCheckWaitMS`**: 无工作时倒计时投票间隔
-   **`lockTimeoutMS`**: 任务在被考虑卡住之前可以锁定的最大时间
-   **`unlockJobsIntervalMS`**: 检查和解锁被困工作的次数
-   **`refreshQueuesIntervalMS`**: 多久刷新要消耗的队列列表
-   **`exclude`**: 跳过处理的队列名称矩阵
-   **`queueSelectionStrategy`**:工人如何选择哪一个可行动队列来接下一份工作. 见 [队列选择策略](https://npmcn.com/package/%40clipboard-health/mongo-jobs/#readme-queue-selection-strategies)

#### 队列选择策略

需要数个队列的工人,每次货币槽解冻时必须选择一个.

-   **`random`** (默认)随机统一选择。 节奏 _尝试_ 分布均匀,但货币不是:一个插槽会一直被占用,直到它所完成的工作,所以一排长跑的工作会堆积插槽并会饿死较短的排子.
-   **`leastInFlight`** 在目前该工人从事的工作最少的队列中统一选择。 这把任何一个队列都限制在工人的货币的公平份额上,因此积压的慢工不能挤出快工队. 作为交换,排队混合的慢而快的工作不再为慢而赚取额外的货币.

这两项战略都符合工作需要:一个工人拥有一个自由的档位,任何可以行动的排队总是要工作。 下方的公平是每个队列,而不是每个任务——一个排有一百万个待办工作的队列得到与一个排队相同的份额.

### 连续任务

使用 cron 表达式安排经常性任务 :

```ts
import { BackgroundJobs } from "@clipboard-health/mongo-jobs";

import { DailyReportJob } from "./jobs/dailyReportJob";

const backgroundJobs = new BackgroundJobs();

// Register a cron job
await backgroundJobs.registerCron(DailyReportJob, {
  // Group assignment (same as regular registration)
  group: "reports",

  // Unique name for this schedule
  scheduleName: "daily-report",

  // Cron expression (standard 5-field format)
  cronExpression: "0 9 * * *", // Every day at 9 AM

  // Optional: timezone for cron evaluation (default: "utc")
  timeZone: "America/New_York",

  // Data to pass to each job execution
  data: { reportType: "daily" },
});
```

```ts
import { backgroundJobs } from "./jobsRegistry";

// Remove a cron schedule and its pending jobs
await backgroundJobs.removeCron("daily-report");
```

#### Cron 列表细节

-   使用标准 5- 字段克龙表达式 : `minute hour day month weekday`
-   在当前任务完成后自动排入下一个任务
-   更新到 cron 调度表会自动取消待完成的任务并重新安排时间
-   失败的 cron 任务按照 `maxAttempts`,但下一个预定的工作仍然会排队
-   每次预定执行都是独特的任务

#### 删除 cron 时间表

**重要**: 当您注册一个 cron 时间表时, 它会持续在数据库中 。 即使你从代码中删除了时间表注册,它将继续执行. 要停止计算表,必须使用 `removeCron` API : (英语)

```ts
await backgroundJobs.removeCron("daily-report");
```

这将:

-   从数据库中删除时间表
-   取消此调度所创建的所有待办工作
-   防止未来工作被安排

### 工作独有性

防止重复的任务被同步排列或运行 :

```ts
import { ProcessUserJob } from "./jobs/processUserJob";
import { backgroundJobs } from "./jobsRegistry";

// Simple uniqueness - single unique key for both enqueued and running
await backgroundJobs.enqueue(
  ProcessUserJob,
  { userId: "123" },
  {
    unique: "process-user-123",
  },
);
```

#### 高级独有性

有可能有单独排列和运行的密钥. 工作完成后,图书馆会 确保我们无法再排队 但一旦它开始运行 它就切换到它运行的键 我们可以再排一个 等待被处决的人 直到第一个完成.

一个可以有用的例子就是重新计算某种缓存. 我们不想再排队了 一个非运行任务,以不爆炸数的被排列的工作。 但一旦它开始运行,还有另一个 触发器可能会需要缓存重新计算,我们希望再安排一个来进行再次重新计算,甚至 如果有一个运行, 因为我们不知道目前的重新计算是否会包括最新的变化.

```ts
import { ProcessUserJob } from "./jobs/processUserJob";
import { backgroundJobs } from "./jobsRegistry";

// Advanced uniqueness - separate keys for enqueued vs running states
await backgroundJobs.enqueue(
  ProcessUserJob,
  { userId: "123" },
  {
    unique: {
      // Only one enqueued job per user
      enqueuedKey: "process-user-123",

      // Only one running job per user
      runningKey: "process-user-123-running",
    },
  },
);
```

```ts
import { SendEmailJob } from "./jobs/sendEmailJob";
import { backgroundJobs } from "./jobsRegistry";

// Example: Allow multiple enqueued but only one running
await backgroundJobs.enqueue(
  SendEmailJob,
  { userId: "123", emailType: "welcome" },
  {
    unique: {
      enqueuedKey: undefined, // Allow multiple enqueued emails
      runningKey: "send-email-123", // But only one sending at a time
    },
  },
);
```

#### 独特行为

-   **排列独有性**:防止重复的任务被添加到队列中. 如果有同样的工作 `enqueuedKey` 已存在且尚未启动, 新的顺序返回 `undefined`
-   **运行独有性**:当工作开始时,其独特的关键转换从 `enqueuedKey` 改为 `runningKey`。这可以防止多个实例同时运行
-   如果检测到重复的独有密钥, 操作静默失败并返回 `undefined`
-   独一性通过MongoDB独一索引执行 `uniqueKey` 字段
-   Cron 任务自动使用基于调度名称和时间戳的独有密钥

## 观察性

### 计量

库默认使用 StatsD 自动报告度量衡 。 每个队列每60秒报告一次计量,包括:

-   **`background_jobs.queue.scheduled`** - 计划今后执行的工作数目
-   **`background_jobs.queue.pending`** - 准备处理的工作数目
-   **`background_jobs.queue.created`** - 工作总数(排定的+待定)
-   **`background_jobs.queue.failed`** - 用尽所有尝试的工作数量
-   **`background_jobs.queue.retry`** - 复试后反增薪
-   **`background_jobs.queue.expired`** - 工作锁到期时反增薪(固定工作)
-   **`background_jobs.queue.delay`** - 延迟执行的时间衡量标准(间隔时间) `nextRunAt` 执行)

所有衡量标准都标有 `queue` 以识别该参数属于哪个队列.

#### 自定义度量衡记录器

您可以通过执行 `MetricsReporter` 接口 :

```ts
import { BackgroundJobs, type MetricsReporter } from "@clipboard-health/mongo-jobs";

class CustomMetricsReporter implements MetricsReporter {
  gauge(name: string, value: number, tags: Record<string, string>): void {
    // Report gauge metric
    console.log(`Gauge: ${name} = ${value}`, tags);
  }

  increment(name: string, tags: Record<string, string>): void {
    // Report counter increment
    console.log(`Increment: ${name}`, tags);
  }

  timing(name: string, value: number | Date, tags: Record<string, string>): void {
    // Report timing metric
    console.log(`Timing: ${name} = ${value}`, tags);
  }
}

const backgroundJobs = new BackgroundJobs({
  metricsReporter: new CustomMetricsReporter(),
});
```

#### StatsD 配置

默认的度量衡记录员使用 `hot-shots` StatsD客户端. 您可以通过传递选项来配置 :

```ts
import { BackgroundJobs, defaultMetricsReporter } from "@clipboard-health/mongo-jobs";

const backgroundJobs = new BackgroundJobs({
  metricsReporter: defaultMetricsReporter({
    host: "localhost",
    port: 8125,
    globalTags: { env: "production" },
  }),
});
```

### OpenTeleometry 跟踪

库提供内置的OpenTeleometry分布式追踪支持. 跟踪会自动为工作排队(生产者)和执行(消费者)创建,这样您就可以通过分布式系统跟踪工作.

#### 追踪跨度

创造了三种跨度:

1.  **制片人间距** (单位:千美元)`background-jobs.producer`\- 工作排队时创建
    
    -   请: `PRODUCER`
    -   属性包括: 消息系统, 操作, 目的地( 手取名), 队列名称
2.  **消费范围** (单位:千美元)`background-jobs.consumer`\- 工作执行时创建
    
    -   请: `CONSUMER`
    -   与生产者之间建立联系,用于分布式追踪
    -   属性包括:消息ID、处理器名称、队列、尝试计数、时间戳
3.  **内部跨度** (单位:千美元)`background-jobs.internals`\- 为内部业务创建
    
    -   请: `INTERNAL`
    -   用于打工,上报指标等业务.

#### 设置 OpenTelection

为了能够追踪, 请在您的应用程序中配置 OpenTeleometry SDK :

```ts
import { NodeSDK } from "@opentelemetry/sdk-node";
import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter({
    url: "http://localhost:4318/v1/traces",
  }),
  instrumentations: [getNodeAutoInstrumentations()],
});

sdk.start();
```

#### 分布式追踪

当一个任务被选中时, 微量上下文会通过 `_traceHeaders` 字段。 当任务执行时,会提取出这个上下文,将消费者的跨度与生产者的跨度联系起来,从而实现端到端的跟踪能见度.

```text
HTTP Request → Enqueue Job (Producer Span)
                    ↓
              [Job in Queue]
                    ↓
              Execute Job (Consumer Span) → Your Handler
```

## 许可证

麻省理工学院

## 常见问题

### @clipboard-health/mongo-jobs 是什么？

MongoDB 驱动的背景工作 \.

### 如何安装 @clipboard-health/mongo-jobs？

在项目目录运行 npm install @clipboard-health/mongo-jobs，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：mongodb（\>=6\.9）、mongoose（\>=8\.7）、@opentelemetry/api（\>=1\.0）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。

### @clipboard-health/mongo-jobs 支持 TypeScript 吗？

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

### @clipboard-health/mongo-jobs 支持 ESM 还是 CommonJS？

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

### @clipboard-health/mongo-jobs 可以商用吗？

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

### @clipboard-health/mongo-jobs 的下载量有多少？

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

### @clipboard-health/mongo-jobs 还在维护吗？

最新版本 2\.8\.2 发布于 2026/10/22（19 天前）；至今共发布 366 个版本，其中过去 12 个月发布了 344 个；首次发布于 2025/10/15。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。