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

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

特征

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

安装

npm install @clipboard-health/mongo-jobs

快速启动

1. 界定工作管理人

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

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 实例并登记您的处理器到组:

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

import { WelcomeEmailJob } from "./welcomeEmailJob";

const backgroundJobs = new BackgroundJobs();

backgroundJobs.register(WelcomeEmailJob, "emails");

export { backgroundJobs };

3. 安排工作

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

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

await backgroundJobs.enqueue(WelcomeEmailJob, {
  userId: "123",
  email: "[email protected]",
});

4. 启动工人

从队列开始处理任务 :

import { backgroundJobs } from "./jobsRegistry";

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

使用量

创造就业

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

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 实例并分配给处理组:

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");

团体允许您:

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

有依赖性的工作

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

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");

有依赖关系的示例工作 :

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 参数)

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

// 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()。 。 。 .

排序任务

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

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

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

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,
  },
);

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会话

创办工人

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

import { backgroundJobs } from "./jobsRegistry";

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

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",
});

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:工人如何选择哪一个可行动队列来接下一份工作. 见 队列选择策略

队列选择策略

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

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

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

连续任务

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

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" },
});

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 : (英语)

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

这将:

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

工作独有性

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

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",
  },
);

高级独有性

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

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

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",
    },
  },
);

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 接口 :

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客户端. 您可以通过传递选项来配置 :

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 :

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 字段。 当任务执行时,会提取出这个上下文,将消费者的跨度与生产者的跨度联系起来,从而实现端到端的跟踪能见度.

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

许可证

麻省理工学院