极简定时任务系统用 Cron 数据库锁实现分布式调度的工程化方案一、两个实例同时跑定时任务数据重复了——单机 cron 在分布式环境中的失效定时任务在单机环境下很简单crontab 写一条规则到点执行。但当服务部署在多台机器上时同样的 crontab 会让每个实例都在相同时间触发相同的任务——数据重复处理、资源竞争、甚至造成幂等性失效。分布式定时任务系统如 XXL-JOB、Quartz、Airflow功能强大但引入的复杂度和运维成本对于小型项目来说是过度的。大多数项目需要的只是一套防止任务重复执行、支持失败重试、有基本监控的轻量方案。用 crontab 做时间触发、用数据库做分布式锁、用应用代码做任务编排——这套组合已经能覆盖 90% 的场景。二、极简分布式定时任务的三层架构graph TD A[Cron 表达式br/秒级触发] -- B[任务调度器] B -- C{获取分布式锁} C --|成功| D[执行任务] C --|失败(其他实例已持有锁)| E[跳过本次执行] D -- F{执行结果} F --|成功| G[更新任务状态br/记录日志] F --|失败| H{重试次数 最大值?} H --|是| I[等待退避时间br/重新执行] H --|否| J[标记失败br/发送告警] G -- K[释放锁] J -- K核心设计思想每一台机器上都有完整的 cron 逻辑但在执行任务前通过数据库行锁SELECT ... FOR UPDATE竞争执行权。获胜者拥有锁、执行任务失败者跳过本次触发。这把分布式调度降级为分布式竞争避免了中心化调度器的单点问题。三、生产级分布式定时任务的完整实现数据库表设计-- 任务定义表——定义了有哪些任务 CREATE TABLE scheduled_tasks ( id BIGSERIAL PRIMARY KEY, name VARCHAR(128) NOT NULL UNIQUE, cron_expr VARCHAR(64) NOT NULL, -- cron 表达式 handler VARCHAR(256) NOT NULL, -- 任务处理函数标识 enabled BOOLEAN NOT NULL DEFAULT true, timeout_seconds INT NOT NULL DEFAULT 300, -- 单次执行超时 max_retries INT NOT NULL DEFAULT 0, -- 最大重试次数 retry_delay_seconds INT NOT NULL DEFAULT 60, -- 重试间隔 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); -- 任务执行记录表——记录每次执行 CREATE TABLE task_executions ( id BIGSERIAL PRIMARY KEY, task_name VARCHAR(128) NOT NULL, instance_id VARCHAR(64) NOT NULL, -- 执行实例标识 status VARCHAR(16) NOT NULL DEFAULT pending, -- pending/running/success/failed started_at TIMESTAMPTZ, finished_at TIMESTAMPTZ, error_message TEXT, retry_count INT NOT NULL DEFAULT 0, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); -- 分布式锁表——使用行锁实现竞争 CREATE TABLE task_locks ( task_name VARCHAR(128) PRIMARY KEY, instance_id VARCHAR(64) NOT NULL, locked_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), expires_at TIMESTAMPTZ NOT NULL -- 锁过期时间防死锁 ); CREATE INDEX idx_executions_task_status ON task_executions(task_name, status); CREATE INDEX idx_executions_created ON task_executions(created_at);Go 调度器实现// scheduler/scheduler.go — 极简分布式调度器 package scheduler import ( context database/sql fmt log os sync time github.com/robfig/cron/v3 ) // TaskHandler 任务处理函数签名 type TaskHandler func(ctx context.Context) error type Scheduler struct { db *sql.DB cron *cron.Cron instanceID string handlers map[string]TaskHandler entries map[string]cron.EntryID mu sync.RWMutex lockTimeout time.Duration logger *log.Logger } func NewScheduler(db *sql.DB, lockTimeout time.Duration) *Scheduler { hostname, _ : os.Hostname() return Scheduler{ db: db, cron: cron.New(cron.WithSeconds()), // 支持秒级 cron instanceID: fmt.Sprintf(%s-%d, hostname, os.Getpid()), handlers: make(map[string]TaskHandler), entries: make(map[string]cron.EntryID), lockTimeout: lockTimeout, logger: log.New(os.Stdout, [scheduler] , log.LstdFlags), } } func (s *Scheduler) Register(name, cronExpr string, handler TaskHandler) error { s.mu.Lock() defer s.mu.Unlock() if _, exists : s.handlers[name]; exists { return fmt.Errorf(task %s already registered, name) } s.handlers[name] handler entryID, err : s.cron.AddFunc(cronExpr, func() { s.executeTask(name) }) if err ! nil { return fmt.Errorf(add cron for %s: %w, name, err) } s.entries[name] entryID s.logger.Printf(Registered task: %s (cron: %s), name, cronExpr) return nil } func (s *Scheduler) Start() { s.logger.Printf(Scheduler starting (instance: %s)..., s.instanceID) // 清理本实例遗留的过期锁进程重启场景 s.cleanupStaleLocks() s.cron.Start() s.logger.Println(Scheduler started) } func (s *Scheduler) Stop() context.Context { s.logger.Println(Scheduler stopping...) ctx : s.cron.Stop() // 释放本实例持有的所有锁 s.releaseAllLocks() s.logger.Println(Scheduler stopped) return ctx } func (s *Scheduler) executeTask(name string) { // Step 1: 尝试获取分布式锁 locked, err : s.acquireLock(name) if err ! nil { s.logger.Printf(Failed to acquire lock for %s: %v, name, err) return } if !locked { // 其他实例正在执行跳过 return } defer s.releaseLock(name) // Step 2: 记录执行 executionID, err : s.recordExecution(name) if err ! nil { s.logger.Printf(Failed to record execution for %s: %v, name, err) return } // Step 3: 执行任务带超时和重试 handler, ok : s.handlers[name] if !ok { s.logger.Printf(Handler not found for task: %s, name) return } ctx, cancel : context.WithTimeout(context.Background(), s.lockTimeout) defer cancel() var lastErr error maxRetries : 3 // 可从数据库读取 for attempt : 0; attempt maxRetries; attempt { if attempt 0 { s.logger.Printf(Retry %d/%d for task %s, attempt, maxRetries, name) select { case -ctx.Done(): s.markExecutionFailed(executionID, context cancelled) return case -time.After(time.Duration(attempt) * 30 * time.Second): } } // 刷新锁——长时间任务可能超过锁超时 if err : s.refreshLock(name); err ! nil { s.logger.Printf(Lock refresh failed for %s: %v, name, err) s.markExecutionFailed(executionID, lock refresh failed) return } start : time.Now() err : handler(ctx) elapsed : time.Since(start) if err nil { s.markExecutionSuccess(executionID, elapsed) s.logger.Printf(Task %s completed in %v, name, elapsed) return } lastErr err s.logger.Printf(Task %s attempt %d failed: %v, name, attempt, err) } s.markExecutionFailed(executionID, fmt.Sprintf(all retries exhausted: %v, lastErr)) } // acquireLock 通过数据库行锁竞争执行权 func (s *Scheduler) acquireLock(name string) (bool, error) { tx, err : s.db.Begin() if err ! nil { return false, err } defer tx.Rollback() expiresAt : time.Now().Add(s.lockTimeout) // INSERT ... ON CONFLICT 原子操作 result, err : tx.Exec( INSERT INTO task_locks (task_name, instance_id, locked_at, expires_at) VALUES ($1, $2, NOW(), $3) ON CONFLICT (task_name) DO UPDATE SET instance_id $2, locked_at NOW(), expires_at $3 WHERE task_locks.expires_at NOW() -- 只有锁已过期时才能抢占 , name, s.instanceID, expiresAt) if err ! nil { return false, fmt.Errorf(acquire lock SQL: %w, err) } rowsAffected, _ : result.RowsAffected() if rowsAffected 0 { // 锁被其他实例持有且未过期 return false, tx.Commit() // 提交事务以释放行锁 } return true, tx.Commit() } func (s *Scheduler) releaseLock(name string) { _, err : s.db.Exec( DELETE FROM task_locks WHERE task_name $1 AND instance_id $2 , name, s.instanceID) if err ! nil { s.logger.Printf(Failed to release lock for %s: %v, name, err) } } func (s *Scheduler) refreshLock(name string) error { expiresAt : time.Now().Add(s.lockTimeout) _, err : s.db.Exec( UPDATE task_locks SET expires_at $1 WHERE task_name $2 AND instance_id $3 , expiresAt, name, s.instanceID) return err } func (s *Scheduler) cleanupStaleLocks() { // 删除过期超过 1 小时的锁极端情况下用于人工清理 _, err : s.db.Exec( DELETE FROM task_locks WHERE expires_at NOW() - INTERVAL 1 hour ) if err ! nil { s.logger.Printf(Failed to cleanup stale locks: %v, err) } // 删除本实例进程重启后遗留的锁 _, err s.db.Exec( DELETE FROM task_locks WHERE instance_id $1 , s.instanceID) if err ! nil { s.logger.Printf(Failed to cleanup own locks: %v, err) } } func (s *Scheduler) releaseAllLocks() { _, err : s.db.Exec( DELETE FROM task_locks WHERE instance_id $1 , s.instanceID) if err ! nil { s.logger.Printf(Failed to release all locks: %v, err) } } func (s *Scheduler) recordExecution(name string) (int64, error) { var id int64 err : s.db.QueryRow( INSERT INTO task_executions (task_name, instance_id, status) VALUES ($1, $2, running) RETURNING id , name, s.instanceID).Scan(id) return id, err } func (s *Scheduler) markExecutionSuccess(id int64, elapsed time.Duration) { s.db.Exec( UPDATE task_executions SET status success, finished_at NOW() WHERE id $1 , id) } func (s *Scheduler) markExecutionFailed(id int64, errMsg string) { s.db.Exec( UPDATE task_executions SET status failed, finished_at NOW(), error_message $2 WHERE id $1 , id, errMsg) }使用示例func main() { db, _ : sql.Open(postgres, os.Getenv(DATABASE_URL)) scheduler : NewScheduler(db, 5*time.Minute) // 注册每日数据清理任务 scheduler.Register(daily_cleanup, 0 0 3 * * *, func(ctx context.Context) error { // 清理 30 天前的数据 _, err : db.ExecContext(ctx, DELETE FROM task_executions WHERE created_at NOW() - INTERVAL 30 days ) return err }) // 注册每 5 分钟数据同步任务 scheduler.Register(data_sync, 0 */5 * * * *, func(ctx context.Context) error { // 同步外部数据 return syncExternalData(ctx, db) }) scheduler.Start() defer scheduler.Stop() // 等待信号... select {} }四、方案边界钟漂、锁超时与性能上限时钟漂移。各机器的系统时钟不可能绝对一致。如果实例 A 的时钟比实例 B 快 5 秒同样*/5 *的 cron 表达式会让 A 先触发。这是数据库行锁方案的优势——谁先抢到锁谁执行时钟差异不影响正确性。锁超时设置不当。lockTimeout必须大于任务的最长执行时间。如果任务实际需要 10 分钟但锁超时设为 5 分钟锁会在任务执行中被释放其他实例可能抢到锁导致重复执行。解决方案在长时间任务中周期性调用refreshLock续期。性能上限。基于数据库行锁的方案受限于数据库的并发能力。如果任务数量超过数百个且都是高频触发秒级PostgreSQL 的行锁竞争会带来可观测的性能下降。这种规模下建议迁移到 Redis 锁或专业的分布式调度器。Cron 的秒级限制。标准 cron 最小粒度是分钟。robfig/cron通过cron.WithSeconds()支持秒级触发但秒级触发的任务应控制在个位数——数据库锁在高频竞争下的开销会直线上升。五、总结Cron 数据库锁的方案将分布式调度简化为分布式锁竞争。优点是不引入新的中间件、不依赖中心化调度器、每个实例对等部署。核心原则谁抢到锁谁执行锁必须有过期时间和续期机制。落地方案先建task_locks和task_executions两张表封装acquireLock/refreshLock/releaseLock三个方法再围绕它们构建任务注册和执行的框架。当任务数量超过 50 个或执行频率达到秒级时再评估是否迁移到专业的调度中间件。对于绝大多数中小项目这个不到 200 行的方案已经足够。