首页 / 科技 / 生产级分布式 Job Scheduler 自研实录：基于 Raft + etcd 搞定漏触发、重复触发、脑裂三大顽疾

生产级分布式 Job Scheduler 自研实录：基于 Raft + etcd 搞定漏触发、重复触发、脑裂三大顽疾

摸鱼不慌
摸鱼不慌管理员
0
作者定位：十年后端 / 分布式方向，踩过 cron、Quartz、xxl-job、Elastic-Job 一堆坑，本文是某金融结算场景下的自研落地复盘。代码可直接跑，Go 1.22+ / etcd 3.5+。

一、开篇先把事故摆出来（为什么要自研）

2025 年 Q3，我们支付对账链路的"每日 00:05 生成渠道报表"任务，两周内出了三次问题：
时间
现象
根因
8/12
报表没生成，运营炸锅
单机 cron 节点宕机，漏触发
8/19
两家渠道各扣了一次手续费
双实例共享 cron 配置，重复触发
8/24
网络分区后两节点同时抢锁，数据写花
Redis SETNX 方案，脑裂
当时团队第一反应是"换 xxl-job"，但我们的场景有两个特殊点：

  • 任务必须严格幂等 + 精确一次（扣款类不能靠"反正幂等"敷衍）


  • 集群规模小（3～5 节点），不想再引一个 Admin 控制台 + 独立 DB


于是决定自研一个 轻量分布式 Scheduler：Raft 选主（etcd 自带）+ etcd Lease 抢占 + 任务状态机原子化。下面给的是生产缩略版代码，非玩具实现。

二、架构总览

┌─────────────────────────────────────┐
│         Scheduler Cluster (3 nodes) │
│                                     │
│  ┌──────────┐   ┌──────────┐        │
│  │ Candidate│   │ Candidate│        │
│  └────┬─────┘   └────┬─────┘        │
│       │     Raft      │              │
│  ┌────▼──────────────▼────┐          │
│  │     Leader (唯一)      │ ← etcd election
│  │  ┌─────────────────┐  │
│  │  │ Schedule Loop    │  │  PENDING→LOCKED→RUNNING→SUCCESS/FAILED
│  │  │ Job Registry     │  │
│  │  │ State Machine    │  │
│  │  └─────────────────┘  │
│  └──────┬────────────────┘
│         │ etcd txn (lease)
│  ┌──────▼────────────────┐
│  │   Worker Pool (local) │
│  └──────┬────────────────┘
│         │
│  ┌──────▼────────────────┐
│  │  Business Handler     │  ← 用户自己保证幂等
│  └───────────────────────┘
└─────────────────────────────────────┘
关键选型理由（Why not Redis / Why not ZooKeeper）：

  • etcd Lease + Revision 天然支持"租约过期 = 节点掉线"，比 Redis SETNX + EXPIRE 少一次竞态窗口


  • etcd txn（事务） 可以做 PENDING→LOCKED的 CAS，避免两个 Worker 同时抢同一 Job


  • 不想再维护 ZK 的 JVM 堆，etcd 单二进制 + Go 客户端更顺手


三、核心数据模型

// model/job.gopackage modelimport (	"time")type Job struct {
	ID       string            `json:"id"`        // 全局唯一，幂等注册用
	Name     string            `json:"name"`
	CronExpr string            `json:"cron_expr"` // "0 5 0 * * *" 六字段
	Payload  map[string]string `json:"payload"`
	Version  int64             `json:"version"`   // etcd ModRevision，乐观锁
	TTLSec   int64             `json:"ttl_sec"`   // 单次执行超时}// JobRun 记录一次执行生命周期type JobRun struct {
	RunID     string    `json:"run_id"`
	JobID     string    `json:"job_id"`
	NodeID    string    `json:"node_id"`
	Status    JobStatus `json:"status"` // PENDING / LOCKED / RUNNING / SUCCESS / FAILED
	StartedAt time.Time `json:"started_at"`
	EndedAt   time.Time `json:"ended_at"`
	ErrMsg    string    `json:"err_msg,omitempty"`}type JobStatus stringconst (
	StatusPending JobStatus = "PENDING"
	StatusLocked  JobStatus = "LOCKED"
	StatusRunning JobStatus = "RUNNING"
	StatusSuccess JobStatus = "SUCCESS"
	StatusFailed  JobStatus = "FAILED")
💡 设计取舍：Job 和 JobRun 分开存，Job 是"模板"，JobRun 是"每次触发实例"。这样重试、追溯、对账都有据可查。很多开源调度器把这俩混在一起，后期运维很痛苦。

四、调度器核心（Leader 独占执行）

// scheduler/scheduler.gopackage schedulerimport (	"context"
	"fmt"
	"log"
	"sync"
	"time"

	"go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency")type Config struct {
	EtcdEndpoints []string
	NodeID        string
	Namespace     string // e.g. "/scheduler/my-cluster"}type Scheduler struct {
	cfg       Config
	cli       *clientv3.Client
	session   *concurrency.Session
	election  *concurrency.Election
	jobStore  JobStore
	mu        sync.Mutex
	running   bool
	cancel    context.CancelFunc
}func New(cfg Config) (*Scheduler, error) {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:   cfg.EtcdEndpoints,
		DialTimeout: 5 * time.Second,
	})	if err != nil {		return nil, err
	}
	sess, err := concurrency.NewSession(cli,
		concurrency.WithTTL(10),          // lease 10s
		concurrency.WithContext(context.Background()),
	)	if err != nil {		return nil, err
	}	return &Scheduler{
		cfg:      cfg,
		cli:      cli,
		session:  sess,
		election: concurrency.NewElection(sess, cfg.Namespace+"/election"),
		jobStore: NewEtcdJobStore(cli, cfg.Namespace),
	}, nil}// Run 阻塞运行，自动参与选主func (s *Scheduler) Run() error {
	log.Printf("[%s] joining election...\n", s.cfg.NodeID)	// Campaign 是阻塞的，当选 Leader 后走 Elected 回调
	return s.election.Campaign(context.Background(), s.cfg.NodeID)
}
选主成功后，Leader 启动调度循环：
// 在 Campaign 的 Elected ctx 里调用func (s *Scheduler) serveAsLeader(ctx context.Context) {
	log.Println(">>> I am Leader now")
	ticker := time.NewTicker(1 * time.Second)	defer ticker.Stop()	for {		select {		case <-ctx.Done():
			log.Println("<<< Lost leadership")			return
		case <-ticker.C:
			s.tick(ctx)
		}
	}
}func (s *Scheduler) tick(ctx context.Context) {
	jobs, err := s.jobStore.ListEnabled(ctx)	if err != nil {
		log.Printf("list jobs err: %v", err)		return
	}
	now := time.Now()	for _, j := range jobs {		if !shouldFire(j.CronExpr, now) {			continue
		}		// 异步触发，不 block 调度循环
		go s.triggerJob(ctx, j)
	}
}
⚠️ 踩坑点：tick里千万别串行执行 Job，否则一个慢 Job 会拖住整个调度周期。这里是"触发"而非"执行"，执行丢给 Worker Pool。

五、Job 触发：etcd txn 保证"精确一次"

这是整套设计最关键的 30 行代码——用 etcd 事务 CAS 抢锁，从根源干掉重复触发。
// scheduler/trigger.gofunc (s *Scheduler) triggerJob(ctx context.Context, j *model.Job) {
	runID := fmt.Sprintf("%s-%d", j.ID, time.Now().Unix())
	run := &model.JobRun{
		RunID:  runID,
		JobID:  j.ID,
		NodeID: s.cfg.NodeID,
		Status: model.StatusLocked,
	}	// 1. 把 JobRun 写成 LOCKED（txn 保证同一 Job 同周期只被一个节点抢到）
	key := fmt.Sprintf("%s/runs/%s", s.cfg.Namespace, runID)
	val, _ := json.Marshal(run)

	txn := s.cli.Txn(ctx).		// If: 该 Job 没有"同一 cron 周期"的 RUNNING/Locked 记录
		If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).
		Then(clientv3.OpPut(key, string(val))).
		Else(clientv3.OpGet(key))

	resp, err := txn.Commit()	if err != nil || !resp.Succeeded {		// 被别的 Leader 抢了，静默退出
		return
	}	// 2. 抢到了，推进到 RUNNING，交给本地 Worker
	run.Status = model.StatusRunning
	run.StartedAt = time.Now()
	s.jobStore.UpdateRun(ctx, run)	// 3. 带超时执行（用 Job 自己的 TTL，兜底 30s）
	execCtx, cancel := context.WithTimeout(ctx,
		time.Duration(max(j.TTLSec, 30))*time.Second)	defer cancel()

	err = s.execute(execCtx, j, run)	if err != nil {
		run.Status = model.StatusFailed
		run.ErrMsg = err.Error()
	} else {
		run.Status = model.StatusSuccess
	}
	run.EndedAt = time.Now()
	s.jobStore.UpdateRun(ctx, run)
}
为什么这套能扛脑裂？

  • 只有 Leader 会 tick，但即使出现双主（etcd 丢多数派才会，概率极低），txn.If(Revision==0)也会让第二个节点抢锁失败


  • 就算 Leader 自己挂了，etcd Lease 10s 过期，新 Leader 起来补扫 LOCKED超过 10s 还没 RUNNING的记录——故障转移 ≤ 15s


六、Worker 执行 + 超时熔断

// worker/pool.gopackage workerimport (	"context"
	"fmt"
	"log"
	"sync"
	"time")type Pool struct {
	sem chan struct{} // 限定并发，防止 Job 爆 Goroutine
	handlers map[string]Handler
	mu sync.RWMutex
}type Handler func(ctx context.Context, j *model.Job, runID string) errorfunc NewPool(maxConcurrency int) *Pool {	return &Pool{
		sem:      make(chan struct{}, maxConcurrency),
		handlers: make(map[string]Handler),
	}
}func (p *Pool) Register(jobName string, h Handler) {
	p.mu.Lock()	defer p.mu.Unlock()
	p.handlers[jobName] = h
}func (p *Pool) Submit(ctx context.Context, j *model.Job, run *model.JobRun) {	go func() {
		p.sem <- struct{}{}        // acquire
		defer func() { <-p.sem }() // release

		p.mu.RLock()
		h, ok := p.handlers[j.Name]
		p.mu.RUnlock()		if !ok {
			log.Printf("no handler for job %s", j.Name)			return
		}

		start := time.Now()
		err := h(ctx, j, run.RunID)
		lat := time.Since(start)		if err != nil {
			log.Printf("JOB_FAIL %s run=%s lat=%v err=%v",
				j.ID, run.RunID, lat, err)
		} else {
			log.Printf("JOB_OK   %s run=%s lat=%v",
				j.ID, run.RunID, lat)
		}
	}()
}
业务侧注册 Handler（幂等由业务自己保证，调度器只保送达）：
// biz/handler.gopool.Register("daily-report-gen", func(ctx context.Context, j *model.Job, runID string) error {	// 用 runID / jobID 做去重键，写 DB 时 INSERT ON CONFLICT
	// 这是"精确一次"的最后一道防线
	return generateDailyReport(ctx, j.Payload["biz_date"])
})

七、可观测性（CSDN 编辑部最爱这一段）

生产系统没 metrics 就是裸奔。我们给调度器挂了 Prometheus：
// metrics/metrics.govar (
	jobTriggerTotal = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: "scheduler",
			Name:      "job_trigger_total",
			Help:      "Total job triggers by status",
		}, []string{"job_id", "status"}) // SUCCESS / FAILED / DEDUP

	jobLatency = promauto.NewHistogramVec(
		prometheus.HistogramOpts{
			Namespace: "scheduler",
			Name:      "job_latency_seconds",
			Buckets:   prometheus.DefBuckets,
		}, []string{"job_id"})

	leaderGauge = promauto.NewGauge(
		prometheus.GaugeOpts{
			Namespace: "scheduler",
			Name:      "is_leader",
			Help:      "1 if current node is leader",
		})
)
在 triggerJob里埋：
if resp.Succeeded {
	jobTriggerTotal.WithLabelValues(j.ID, "fired").Inc()
} else {
	jobTriggerTotal.WithLabelValues(j.ID, "dedup").Inc()
}
Grafana 面板核心看板：

  • scheduler_job_trigger_total{status="dedup"}突然飙高 → 双主嫌疑


  • scheduler_job_latency_seconds{p99}超过 cron 周期的 80% → 下一次可能漏触发


  • scheduler_is_leader频繁翻转 → etcd 网络抖动


八、压测 & 对比数据（自研 vs 传统方案）

环境：3 × 4C8G 节点，etcd 3 节点同机，100 个 Job，cron 密度 1/5s。
指标
cron 单机
Redis SETNX
自研（本文）
漏触发（节点宕机）
必然
必然
0（Leader 15s 内接管）
重复触发（双实例）
必然
低概率
0（etcd txn CAS）
脑裂容忍
❌
❌
✅（Raft 多数派）
故障转移耗时
—
依赖 TTL，通常 30s+
≤ 15s
单节点调度 QPS
~500
~3000
~2400（瓶颈在 etcd txn）
📌 瓶颈诚实说：etcd txn 串行写是调度侧的瓶颈，但我们场景单节点 2400 QPS 已经远超"每日报表"类需求（< 100 QPS）。如果你要 10w+ 高频调度，应该分集群或者上 时间轮 + 批量 txn，那是另一个话题。

九、认知升华：什么时候不该自研

十年老哥掏心窝一句：能不用自研就别自研。下面情况直接用 xxl-job / Airflow / Temporal：

  • 任务类型杂（Shell / SQL / Java / Python 都要）


  • 需要 Web 控制台、权限、依赖 DAG


  • 团队没人力养 etcd 集群


本文方案适合的是：小集群（3～7 节点）+ Go 技术栈 + 任务语义简单但要"精确一次" + 不想再引一个调度 Admin。我们金融链路这个场景，xxl-job 的"分片广播 + 幂等"也能凑，但自研 1200 行 Go 换来的是零外部依赖、故障转移可控、metrics 完全贴合业务——这笔账我们算过，划算。

十、代码结构（给想抄的兄弟）

task-scheduler/
├── main.go              # 启动：etcd 连接 → 选主 → Run()
├── scheduler/
│   ├── scheduler.go     # 选主、tick、serveAsLeader
│   ├── trigger.go       # txn CAS 抢锁（核心 30 行）
│   └── cron.go          # cron 解析（用 robfig/cron/v3）
├── worker/
│   └── pool.go          # 并发池 + Handler 注册
├── store/
│   └── etcdstore.go     # Job / JobRun CRUD
├── metrics/
│   └── metrics.go       # Prometheus 埋点
└── biz/
    └── handler.go       # 业务 Handler 注册示例
完整可跑版本（含 Docker Compose etcd + Makefile + 压测脚本）放 Gitee 了，评论区留"求源码"我私信发链接，避免 CSDN 外链被吞。
参考文献 / 类比阅读

  • etcd concurrency 包源码：clientv3/concurrency/election.go（选主实现比你自己写的稳）


  • Google "Reliable Cron across Data Centers"（对漏触发问题的经典论述）


  • 邓明 go-scheduler 开源实现（思路接近，本文在 txn CAS + metrics 上补了两刀）



本文由 摸鱼不慌 发布，转载请注明出处。
文章链接：https://www.moyubuhuang.com/keji/202607/42668.html
0

相关文章