Node.js 独立产品定时任务设计：从 Cron 到分布式调度

一、独立产品定时任务的技术演进路径

独立产品在不同发展阶段，对定时任务的需求差异显著。创业初期只需简单的数据备份，成长期需要复杂的业务调度，规模化后面临分布式挑战。

技术选型需要匹配当前阶段，避免过度设计。从单机 Cron 到分布式调度系统，存在清晰的技术演进路径。

// 阶段 1：简单 Cron（适合 MVP 阶段） // 使用 node-cron 或 system Cron import cron from 'node-cron'; // 每天凌晨 2 点执行数据备份 cron.schedule('0 2 * * *', async () => { console.log('Starting daily backup...'); try { await backupDatabase(); console.log('Backup completed'); } catch (error) { console.error('Backup failed:', error); } }); // 存在问题： // 1. 单机故障导致任务丢失 // 2. 无法水平扩展 // 3. 缺乏任务监控和重试机制 // 4. 时间漂移（服务器时间不准确）

各阶段的技术特征对比：

graph LR A[MVP 阶段<br/>单机 Cron] --> B[成长期<br/>任务队列] B --> C[规模化<br/>分布式调度] A -->|问题：单点故障| B B -->|问题：任务冲突| C B --> D[技术选型] D --> E[Bull Queue<br/>Redis 持久化] D --> F[Agenda<br/>MongoDB 持久化] C --> G[分布式锁] C --> H[任务分片] C --> I[故障转移] style A fill:#e1f5fe style B fill:#fff3e0 style C fill:#f3e5f5

二、基于 Redis 的任务队列方案

对于成长期的独立产品，推荐使用基于 Redis 的任务队列（如 Bull Queue）实现定时任务。优势包括：持久化、重试机制、任务优先级、分布式支持。

Bull Queue 的核心概念：

1. Queue：任务队列，负责存储和分发任务
2. Job：具体的任务实例，包含任务数据和执行状态
3. Worker：任务执行器，从队列中获取任务并执行
4. Scheduler：调度器，负责定时任务的触发

场景描述：Bull Queue 的 Repeatable Job 有一个容易被忽视的特性：当同一个 Cron 任务被多次add时，Bull 不会去重，而是创建多个独立的重复任务。我们发现线上数据备份任务在执行两次，排查后发现是 PM2 集群模式下，每个进程的启动脚本都调用了addCronJob，导致 4 个进程 = 4 个重复任务，数据库每晚被备份 4 次。

// 单实例环境下，用 jobId 去重 await queue.add(taskType, data, { repeat: { cron: '0 2 * * *' }, jobId: `daily-backup-${taskType}` // 固定 jobId 阻止重复创建 });

但在多实例部署时，上述方法只能保证同一队列内不去重，不能防止不同实例的队列都创建了相同的 Repeatable Job。最终方案是：把 Cron 任务的注册放在独立的管理进程或初始化脚本中，而不是放在每个业务进程的启动代码里。

// 使用 Bull Queue 实现定时任务 // services/taskQueue.ts import Queue, { Job } from 'bull'; import Redis from 'ioredis'; export interface TaskData { type: string; payload: Record<string, any>; metadata?: { userId?: string; organizationId?: string; }; } export class TaskQueueService { private queues: Map<string, Queue> = new Map(); private redis: Redis; constructor(redisUrl: string) { this.redis = new Redis(redisUrl, { retryStrategy: (times) => Math.min(times * 50, 2000) }); } /** * 创建或获取队列 */ getQueue(name: string): Queue { if (!this.queues.has(name)) { const queue = new Queue(name, { redis: this.redis, defaultJobOptions: { removeOnComplete: 100, // 保留最近 100 个完成的任务 removeOnFail: 500, // 保留最近 500 个失败的任务 attempts: 3, // 失败重试次数 backoff: { type: 'exponential', delay: 5000 // 重试延迟（毫秒） } } }); // 全局错误处理 queue.on('error', (error) => { console.error(`Queue ${name} error:`, error); }); queue.on('failed', (job, error) => { console.error(`Job ${job.id} failed:`, error); }); this.queues.set(name, queue); } return this.queues.get(name)!; } /** * 添加定时任务（使用 Cron 表达式） */ async addCronJob( queueName: string, taskType: string, payload: Record<string, any>, cronExpression: string, options?: { tz?: string; // 时区 startDate?: Date; // 开始时间 endDate?: Date; // 结束时间 } ): Promise<Job> { const queue = this.getQueue(queueName); return queue.add( taskType, { type: taskType, payload }, { repeat: { cron: cronExpression, tz: options?.tz || 'Asia/Shanghai', startDate: options?.startDate, endDate: options?.endDate } } ); } /** * 添加延迟任务（一次性执行） */ async addDelayedJob( queueName: string, taskType: string, payload: Record<string, any>, delayMs: number ): Promise<Job> { const queue = this.getQueue(queueName); return queue.add( taskType, { type: taskType, payload }, { delay: delayMs } ); } /** * 注册任务处理器 */ registerProcessor( queueName: string, taskType: string, processor: (job: Job<TaskData>) => Promise<void> ): void { const queue = this.getQueue(queueName); queue.process(taskType, async (job) => { const startTime = Date.now(); console.log(`Processing job ${job.id} (${taskType})...`); try { await processor(job); const duration = Date.now() - startTime; console.log(`Job ${job.id} completed in ${duration}ms`); } catch (error) { console.error(`Job ${job.id} failed after ${Date.now() - startTime}ms:`, error); throw error; // 重新抛出错误，触发重试机制 } }); } /** * 获取队列统计信息 */ async getQueueStats(queueName: string): Promise<{ waiting: number; active: number; completed: number; failed: number; delayed: number; }> { const queue = this.getQueue(queueName); const [waiting, active, completed, failed, delayed] = await Promise.all([ queue.getWaitingCount(), queue.getActiveCount(), queue.getCompletedCount(), queue.getFailedCount(), queue.getDelayedCount() ]); return { waiting, active, completed, failed, delayed }; } /** * 清理已完成的任务 */ async cleanQueue(queueName: string, gracePeriodMs: number = 24 * 60 * 60 * 1000): Promise<void> { const queue = this.getQueue(queueName); await Promise.all([ queue.clean(gracePeriodMs, 'completed'), queue.clean(gracePeriodMs, 'failed') ]); console.log(`Cleaned old jobs from queue: ${queueName}`); } }

实际业务场景示例：

// tasks/userTasks.ts - 用户相关的定时任务 import { TaskQueueService } from '../services/taskQueueService'; export function registerUserTasks(taskService: TaskQueueService): void { // 任务 1：每日用户活跃度统计（凌晨 1 点执行） taskService.registerProcessor('user-tasks', 'daily-activity-stats', async (job) => { const { organizationId } = job.data.payload; // 统计昨天的用户活跃数据 const stats = await calculateDailyActivity(organizationId, 'yesterday'); // 保存统计结果 await saveActivityStats(stats); // 如果活跃度下降超过 20%，发送告警 if (stats.activityDropRate > 0.2) { await sendAlert({ type: 'activity_drop', organizationId, data: stats }); } }); // 添加 Cron 任务 taskService.addCronJob( 'user-tasks', 'daily-activity-stats', { organizationId: 'all' }, '0 1 * * *', // 每天凌晨 1 点 { tz: 'Asia/Shanghai' } ); // 任务 2：定期清理过期邀请链接（每小时执行） taskService.registerProcessor('user-tasks', 'cleanup-invites', async (job) => { const expiredCount = await cleanupExpiredInvites(); console.log(`Cleaned up ${expiredCount} expired invites`); }); taskService.addCronJob( 'user-tasks', 'cleanup-invites', {}, '0 * * * *' // 每小时的第 0 分钟 ); // 任务 3：用户流失预警（每天上午 9 点执行） taskService.registerProcessor('user-tasks', 'churn-prediction', async (job) => { const highRiskUsers = await predictChurnRisk(); for (const user of highRiskUsers) { // 发送个性化挽留邮件 await sendRetentionEmail(user); // 记录预警日志 await logChurnWarning(user); } }); taskService.addCronJob( 'user-tasks', 'churn-prediction', {}, '0 9 * * *', { tz: 'Asia/Shanghai' } ); } // 辅助函数（简化实现） async function calculateDailyActivity(orgId: string, date: string): Promise<any> { // 实现活跃度统计逻辑 return { organizationId: orgId, date, activeUsers: 0, newUsers: 0, activityDropRate: 0 }; } async function saveActivityStats(stats: any): Promise<void> { // 保存统计结果到数据库 console.log('Saving activity stats:', stats); } async function sendAlert(alert: any): Promise<void> { // 发送告警通知 console.log('Sending alert:', alert); } async function cleanupExpiredInvites(): Promise<number> { // 清理过期邀请链接 return 0; } async function predictChurnRisk(): Promise<any[]> { // 预测流失风险 return []; } async function sendRetentionEmail(user: any): Promise<void> { // 发送挽留邮件 console.log('Sending retention email to user:', user); } async function logChurnWarning(user: any): Promise<void> { // 记录预警日志 console.log('Logging churn warning for user:', user); }

三、分布式调度系统的设计

当独立产品发展到多实例部署阶段，需要解决定时任务的分布式执行问题。核心挑战包括：任务重复执行、调度一致性、故障转移。

技术方案：

1. 分布式锁：使用 Redis Redlock 确保同一任务在同一时间只在一个实例上执行
2. 主节点选举：使用 Redis Sentinel 或 etcd 选举调度主节点
3. 任务分片：将大任务拆分为多个子任务，分发到不同实例执行
4. 状态同步：使用 Redis 或数据库同步任务执行状态

场景描述：分布式锁是解决重复执行的核心手段，但锁的 TTL（过期时间）设置需要与任务的预期执行时间匹配。我们曾经设置了一个数据导出任务的锁 TTL 为 60 秒，而实际导出操作耗时约 90 秒——锁 TTL 到期自动释放后，另一个节点抢到锁又启动了一份导出，导致两个节点同时导出数据，用户收到两份通知邮件。教训是：锁的 TTL 至少要设置成任务预期最大耗时的 2 倍，并且在任务执行期间定期续期。

// 分布式任务调度器 // services/distributedScheduler.ts import Redis from 'ioredis'; import Redlock from 'redlock'; export class DistributedScheduler { private redis: Redis; private redlock: Redlock; private nodeId: string; private isLeader: boolean = false; private tasks: Map<string, ScheduledTask> = new Map(); constructor(redisUrl: string, nodeId?: string) { this.redis = new Redis(redisUrl); this.redlock = new Redlock([this.redis], { driftFactor: 0.01, retryCount: 10, retryDelay: 200 }); this.nodeId = nodeId || `${process.pid}-${Date.now()}`; console.log(`Distributed scheduler initialized (Node ID: ${this.nodeId})`); } /** * 启动调度器 */ async start(): Promise<void> { // 1. 尝试获取领导权 await this.electLeader(); // 2. 监听领导权变化 this.startLeaderElection(); // 3. 注册内置任务 this.registerBuiltInTasks(); console.log(`Scheduler started on node ${this.nodeId}`); } /** * 注册定时任务 */ registerTask(task: ScheduledTask): void { if (this.tasks.has(task.name)) { console.warn(`Task ${task.name} already registered, overwriting...`); } this.tasks.set(task.name, task); console.log(`Registered task: ${task.name} (${task.cron})`); } /** * 执行任务（带分布式锁） */ private async executeTaskWithLock(task: ScheduledTask): Promise<void> { const lockKey = `scheduler:lock:${task.name}`; const lockTtl = task.timeout || 60000; // 锁的过期时间（毫秒） try { // 获取分布式锁 const lock = await this.redlock.acquire([lockKey], lockTtl); try { console.log(`Executing task: ${task.name} (Node: ${this.nodeId})`); // 执行任务 await Promise.race([ task.execute(), this.timeoutPromise(task.timeout || 30000) ]); console.log(`Task ${task.name} completed successfully`); } finally { // 释放锁 await lock.unlock(); } } catch (error) { if (error instanceof Error && error.message.includes('LockError')) { console.log(`Task ${task.name} is already running on another node`); } else { console.error(`Task ${task.name} failed:`, error); // 执行失败回调 if (task.onFailure) { await task.onFailure(error); } } } } /** * 领导节点选举 */ private async electLeader(): Promise<void> { const leaderKey = 'scheduler:leader'; const leaderTtl = 10000; // 领导权有效期（毫秒） try { const result = await this.redis.set( leaderKey, this.nodeId, 'NX', 'PX', leaderTtl ); if (result === 'OK') { this.isLeader = true; console.log(`Node ${this.nodeId} elected as leader`); // 续期领导权 this.renewLeadership(leaderKey, leaderTtl); } else { this.isLeader = false; const currentLeader = await this.redis.get(leaderKey); console.log(`Node ${this.nodeId} is follower (Leader: ${currentLeader})`); } } catch (error) { console.error('Leader election failed:', error); this.isLeader = false; } } /** * 续期领导权 */ private async renewLeadership(leaderKey: string, ttl: number): Promise<void> { if (!this.isLeader) return; setTimeout(async () => { try { const result = await this.redis.set( leaderKey, this.nodeId, 'XX', 'PX', ttl ); if (result === 'OK') { this.renewLeadership(leaderKey, ttl); } else { console.log('Leadership expired, re-electing...'); await this.electLeader(); } } catch (error) { console.error('Failed to renew leadership:', error); this.isLeader = false; } }, ttl * 0.6); // 在 TTL 过期前续期 } /** * 启动领导权变化监听 */ private startLeaderElection(): void { // 使用 Redis Pub/Sub 监听领导权变化 const subscriber = this.redis.duplicate(); subscriber.subscribe('scheduler:leader-change'); subscriber.on('message', (channel, message) => { if (channel === 'scheduler:leader-change') { console.log('Leader change detected, re-electing...'); this.electLeader(); } }); // 定期检查领导权 setInterval(() => { if (!this.isLeader) { this.electLeader(); } }, 5000); } /** * 运行调度循环（仅领导节点执行） */ private startScheduling(): void { setInterval(() => { if (!this.isLeader) return; const now = new Date(); for (const [name, task] of this.tasks) { if (this.shouldRunTask(task, now)) { this.executeTaskWithLock(task); } } }, 1000); // 每秒检查一次 } /** * 判断任务是否应该执行 */ private shouldRunTask(task: ScheduledTask, now: Date): boolean { // 简化版 Cron 判断（实际应使用专门的 Cron 解析库） const [minute, hour, dayOfMonth, month, dayOfWeek] = task.cron.split(' '); const matches = ( this.matchesCronField(minute, now.getMinutes()) && this.matchesCronField(hour, now.getHours()) && this.matchesCronField(dayOfMonth, now.getDate()) && this.matchesCronField(month, now.getMonth() + 1) && this.matchesCronField(dayOfWeek, now.getDay()) ); return matches; } private matchesCronField(field: string, value: number): boolean { if (field === '*') return true; if (field.includes(',')) { return field.split(',').map(Number).includes(value); } return Number(field) === value; } private timeoutPromise(ms: number): Promise<never> { return new Promise((_, reject) => { setTimeout(() => reject(new Error('Task timeout')), ms); }); } private registerBuiltInTasks(): void { // 注册内置的健康检查任务 this.registerTask({ name: 'health-check', cron: '*/5 * * * *', // 每 5 分钟 execute: async () => { const health = await this.checkSystemHealth(); console.log('System health:', health); }, timeout: 10000 }); // 注册内置的任务清理任务 this.registerTask({ name: 'cleanup', cron: '0 2 * * *', // 每天凌晨 2 点 execute: async () => { await this.cleanupOldTasks(); }, timeout: 60000 }); } private async checkSystemHealth(): Promise<any> { return { nodeId: this.nodeId, isLeader: this.isLeader, tasksRegistered: this.tasks.size, timestamp: new Date().toISOString() }; } private async cleanupOldTasks(): Promise<void> { console.log('Cleaning up old task records...'); // 实现清理逻辑 } } // 定时任务接口 export interface ScheduledTask { name: string; cron: string; execute: () => Promise<void>; timeout?: number; onFailure?: (error: any) => Promise<void>; }

四、定时任务系统的工程化实践

1. 任务监控与告警：实时追踪任务执行状态，异常时及时通知。

// 任务监控服务 export class TaskMonitor { private metrics: Map<string, TaskMetrics> = new Map(); /** * 记录任务执行 */ recordExecution(taskName: string, success: boolean, durationMs: number): void { let metrics = this.metrics.get(taskName); if (!metrics) { metrics = { taskName, totalExecutions: 0, successfulExecutions: 0, failedExecutions: 0, totalDurationMs: 0, lastExecutionAt: new Date(), lastFailureAt: null }; this.metrics.set(taskName, metrics); } metrics.totalExecutions++; metrics.totalDurationMs += durationMs; metrics.lastExecutionAt = new Date(); if (success) { metrics.successfulExecutions++; } else { metrics.failedExecutions++; metrics.lastFailureAt = new Date(); } // 检查是否需要告警 this.checkAlerts(taskName, metrics); } /** * 检查告警条件 */ private checkAlerts(taskName: string, metrics: TaskMetrics): void { // 告警规则 1：连续失败超过 3 次 if (metrics.failedExecutions >= 3) { this.sendAlert({ level: 'error', taskName, message: `Task ${taskName} failed ${metrics.failedExecutions} times consecutively` }); } // 告警规则 2：执行时间超过阈值 const avgDuration = metrics.totalDurationMs / metrics.totalExecutions; if (avgDuration > 30000) { // 平均超过 30 秒 this.sendAlert({ level: 'warn', taskName, message: `Task ${taskName} average execution time is ${avgDuration}ms` }); } } private sendAlert(alert: { level: string; taskName: string; message: string }): void { console.log(`[${alert.level.toUpperCase()}] ${alert.message}`); // 发送到监控服务（如 Sentry、Datadog） // ... } } interface TaskMetrics { taskName: string; totalExecutions: number; successfulExecutions: number; failedExecutions: number; totalDurationMs: number; lastExecutionAt: Date; lastFailureAt: Date | null; }

2. 任务依赖管理：支持任务间的依赖关系，确保执行顺序。

// 任务依赖管理器 export class TaskDependencyManager { private dependencies: Map<string, string[]> = new Map(); /** * 添加任务依赖 */ addDependency(taskName: string, dependsOn: string[]): void { this.dependencies.set(taskName, dependsOn); } /** * 获取任务的执行顺序（拓扑排序） */ getExecutionOrder(taskNames: string[]): string[] { const result: string[] = []; const visited = new Set<string>(); const visiting = new Set<string>(); const visit = (taskName: string) => { if (visiting.has(taskName)) { throw new Error(`Circular dependency detected at task: ${taskName}`); } if (visited.has(taskName)) { return; } visiting.add(taskName); const deps = this.dependencies.get(taskName) || []; for (const dep of deps) { if (taskNames.includes(dep)) { visit(dep); } } visiting.delete(taskName); visited.add(taskName); result.push(taskName); }; for (const taskName of taskNames) { visit(taskName); } return result; } }

3. 灰度发布支持：新任务先在小范围执行，验证通过后全量上线。

场景描述：灰度发布听起来像是"锦上添花"的功能，直到你上线了一个新的用户流失预警邮件任务，结果配置错误——灰度 10% 变成了 100%，全量用户在一小时内收到了内容错误的预警邮件，客服被打爆。灰度发布的代码实现不难，难的是灰度百分比配置的校验和回滚机制。我们在setGrayConfig中加入了硬上限检查：

setGrayConfig(taskName: string, config: GrayConfig): void { if (config.percentage !== undefined && config.percentage > 20) { console.warn(`Task ${taskName} gray percentage > 20%, requires manual approval`); // 实际项目中应发送审批通知，而不是静默降级 config.percentage = Math.min(config.percentage, 20); } this.grayConfigs.set(taskName, config); }

这个硬限制多次救了我们——团队成员提交灰度配置时不小心把 10% 写成了 100%，上限检查直接拦截了下来。

// 任务灰度管理器 export class TaskGrayReleaseManager { private grayConfigs: Map<string, GrayConfig> = new Map(); /** * 配置灰度规则 */ setGrayConfig(taskName: string, config: GrayConfig): void { this.grayConfigs.set(taskName, config); } /** * 判断是否应该执行任务 */ shouldExecute(taskName: string, context: { userId?: string; organizationId?: string }): boolean { const config = this.grayConfigs.get(taskName); if (!config) { return true; // 没有灰度配置，默认执行 } if (config.enabled === false) { return false; // 任务被禁用 } // 按百分比灰度 if (config.percentage !== undefined) { const hash = this.hashString(context.organizationId || context.userId || ''); const bucket = hash % 100; return bucket < config.percentage; } // 按白名单灰度 if (config.whitelist) { const id = context.organizationId || context.userId; return config.whitelist.includes(id || ''); } return false; } private hashString(str: string): number { let hash = 0; for (let i = 0; i < str.length; i++) { const char = str.charCodeAt(i); hash = ((hash << 5) - hash) + char; hash = hash & hash; } return Math.abs(hash); } } interface GrayConfig { enabled: boolean; percentage?: number; // 灰度百分比（0-100） whitelist?: string[]; // 白名单 startDate?: Date; // 开始时间 endDate?: Date; // 结束时间 }

五、总结

Node.js 独立产品的定时任务设计需要随业务规模演进。从单机 Cron 到分布式调度系统，每个阶段都有合适的技术选型。

关键实践包括：使用 Redis 持久化任务队列、分布式锁避免重复执行、领导选举确保调度一致性、任务监控与告警、依赖管理和灰度发布。

对于资源有限的独立产品，建议从 Bull Queue 等成熟方案起步，避免重复造轮子。当业务规模增长后，再逐步引入分布式调度能力。