mcpField
B
background-jobs-designer

background-jobs-designer

github.com/patricio0312rev开发63272 次安装41 周前更新
代码仓库

关于 background-jobs-designer

background-jobs-designer 是一项帮助开发者设计可靠后台任务处理的技能，示例涉及 BullMQ 和 Celery。内容涵盖重试、幂等处理、错误监控及队列运行实践。

主要功能

  1. 01展示 BullMQ 和 Celery 队列集成
  2. 02设计任务定义与重试策略
  3. 03使用指数退避处理重试
  4. 04规划幂等任务与处理记录
  5. 05监控任务事件和指标

适用场景

  1. 01设计邮件等后台任务队列
  2. 02设置失败任务的重试策略
  3. 03避免重复执行任务造成副作用
  4. 04规划队列深度与失败监控

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/patricio0312rev/skills --skill background-jobs-designer

安装后在 Claude Code 中输入/background-jobs-designer确认已加载。

详细介绍

后台任务设计器

设计可靠的后台任务处理机制，并支持重试和监控。

队列集成

BullMQ (Node.js)：

import { Queue, Worker } from "bullmq";

const emailQueue = new Queue("email", {
  connection: { host: "localhost", port: 6379 },
});

// Add job
await emailQueue.add(
  "send-welcome",
  {
    userId: "123",
    email: "[email protected]",
  },
  {
    attempts: 3,
    backoff: { type: "exponential", delay: 2000 },
  }
);

Celery (Python)：

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task(bind=True, max_retries=3)
def send_email(self, user_id, email):
    try:
        # Send email
        pass
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

任务定义

export interface Job {
  id: string;
  type: string;
  payload: unknown;
  attempts: number;
  maxAttempts: number;
  createdAt: Date;
  processedAt?: Date;
  failedAt?: Date;
  error?: string;
}

export const JOB_TYPES = {
  SEND_EMAIL: "send-email",
  PROCESS_PAYMENT: "process-payment",
  GENERATE_REPORT: "generate-report",
  SYNC_DATA: "sync-data",
} as const;

重试策略

// Exponential backoff
const RETRY_CONFIG = {
  maxAttempts: 5,
  delays: [
    1000, // 1 second
    5000, // 5 seconds
    30000, // 30 seconds
    300000, // 5 minutes
    1800000, // 30 minutes
  ],
};

// Worker with retry
const worker = new Worker("email", async (job) => {
  try {
    await sendEmail(job.data);
  } catch (error) {
    if (job.attemptsMade < RETRY_CONFIG.maxAttempts) {
      throw error; // Will retry
    }
    await handleFailedJob(job, error);
  }
});

幂等任务

// Track processed jobs
export const processJob = async (job: Job) => {
  // Check if already processed
  const processed = await db.query(
    "SELECT 1 FROM processed_jobs WHERE job_id = $1",
    [job.id]
  );

  if (processed.rows.length > 0) {
    console.log("Job already processed");
    return; // Idempotent
  }

  await db.transaction(async (trx) => {
    // Mark as processed
    await trx("processed_jobs").insert({ job_id: job.id });

    // Do work
    await performWork(job, trx);
  });
};

监控

// Job events
worker.on("completed", (job) => {
  metrics.increment("jobs.completed", { type: job.name });
});

worker.on("failed", (job, err) => {
  metrics.increment("jobs.failed", { type: job.name });
  logger.error("Job failed", { jobId: job.id, error: err });
});

worker.on("stalled", (jobId) => {
  metrics.increment("jobs.stalled");
  logger.warn("Job stalled", { jobId });
});

最佳实践

  • 任务应具备幂等性
  • 重试时使用指数退避
  • 设置合理的超时时间
  • 监控队列深度
  • 为失败的任务设置死信队列
  • 记录任务开始和完成情况
  • 妥善处理优雅关闭

输出检查清单

  • 队列设置（Redis/RabbitMQ）
  • 任务类型定义
  • 带退避机制的重试策略
  • 幂等性跟踪
  • 错误处理
  • 监控/指标
  • 死信队列
  • 优雅关闭

评论

支持 Markdown