榜单模型与分布式任务调度

func score(likeCnt int64, utime int64) float64 {
    // utime 是文章发表时间（毫秒）
    hours := float64(time.Since(time.UnixMilli(utime)).Hours())
    return float64(likeCnt-1) / math.Pow(hours+2, 1.5)
}

func (s *RankingService) Start(ctx context.Context) {
    ticker := time.NewTicker(time.Minute) // 每分钟算一次
    defer ticker.Stop()                   // 退出时停止 ticker，防止泄漏
    for {
        select {
        case <-ctx.Done():
            return // 主动退出
        case <-ticker.C:
            _ = s.Rank(ctx) // 触发计算
        }
    }
}

go get github.com/robfig/cron/v3

import "github.com/robfig/cron/v3"

c := cron.New(cron.WithSeconds()) // 启用秒级字段，必须设置
// 添加任务：AddFunc / AddJob 都是线程安全的
id, _ := c.AddJob("*/5 * * * * *", myJob)
c.Start() // 开始调度

// 优雅退出
ctx := c.Stop() // 只是暂停调度，正在运行的任务不会被中断
<-ctx.Done()    // 等待所有任务结束

package ranking

import "container/heap"

// ArticleScore 文章 + 热度分数
type ArticleScore struct {
    Art   domain.Article
    Score float64
}

// priorityQueue 小顶堆，按 Score 升序
type priorityQueue []ArticleScore

func (p priorityQueue) Len() int           { return len(p) }
func (p priorityQueue) Less(i, j int) bool { return p[i].Score < p[j].Score }
func (p priorityQueue) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
func (p *priorityQueue) Push(x any)        { *p = append(*p, x.(ArticleScore)) }
func (p *priorityQueue) Pop() any {
    old := *p
    n := len(old)
    x := old[n-1]
    *p = old[:n-1]
    return x
}

// TopNHeap 限容的小顶堆
type TopNHeap struct {
    h    *priorityQueue
    cap  int
}

func NewTopNHeap(cap int) *TopNHeap {
    pq := make(priorityQueue, 0, cap)
    return &TopNHeap{h: &pq, cap: cap}
}

// Push 入堆：堆满则替换最小值
func (t *TopNHeap) Push(x ArticleScore) {
    if t.h.Len() < t.cap {
        heap.Push(t.h, x)
        return
    }
    // 堆已满，比堆顶小则丢弃
    if x.Score <= (*t.h)[0].Score {
        return
    }
    // 比堆顶大，弹出堆顶，压入新值
    heap.Pop(t.h)
    heap.Push(t.h, x)
}

// PopAll 弹出全部并按降序返回
func (t *TopNHeap) PopAll() []ArticleScore {
    res := make([]ArticleScore, 0, t.h.Len())
    for t.h.Len() > 0 {
        res = append(res, heap.Pop(t.h).(ArticleScore))
    }
    // 弹出是升序，反转成降序
    for i, j := 0, len(res)-1; i < j; i, j = i+1, j-1 {
        res[i], res[j] = res[j], res[i]
    }
    return res
}

package ranking

type RankingService interface {
    RankTopN(ctx context.Context) error
}

type BatchRankingService struct {
    articleSvc ArticleService
    intrSvc    InteractiveService
    batch      int        // 每批大小
    n          int        // TopN
    scoreFunc  func(like int64, utime int64) float64
    cache      RankCache  // Redis 缓存
}

func (s *BatchRankingService) RankTopN(ctx context.Context) error {
    // 步骤 1：初始化限容小顶堆
    heap := NewTopNHeap(s.n)
    offset := 0
    // 只算最近 7 天的文章，老文章不可能上榜
    since := time.Now().AddDate(0, 0, -7).UnixMilli()
    for {
        // 步骤 2：分批拉文章
        arts, err := s.articleSvc.Batch(ctx, offset, s.batch, since)
        if err != nil { return err }
        if len(arts) == 0 { break }
        offset += len(arts)

        // 步骤 3：批量查点赞数
        likeMap, err := s.intrSvc.BatchGetLike(ctx, "article", toIds(arts))
        if err != nil { return err }

        // 步骤 4：计算 score，维护 TopN
        for _, art := range arts {
            score := s.scoreFunc(likeMap[art.Id], art.Utime)
            heap.Push(ArticleScore{Art: art, Score: score})
        }
    }
    // 步骤 5：写入 Redis 缓存
    top := heap.PopAll()
    return s.cache.Save(ctx, top)
}

// 两个关键点：
// 1. 热榜数据通常不存数据库（除非要做大数据分析）
// 2. Redis 过期时间要比计算间隔长，留够重试时间
//    例如每分钟计算，过期时间设 10 分钟

package job

type Job interface {
    Run(ctx context.Context) error
}

type Builder struct {
    p     *prometheus.SummaryVec // 监控
    l     logger.Logger
}

func (b *Builder) Build(name string, j job.Job) cron.Job {
    return cron.FuncJob(func() {
        start := time.Now()
        ctx := context.Background()
        err := j.Run(ctx)
        // 监控执行耗时
        b.p.WithLabelValues(name).Observe(time.Since(start).Seconds())
        if err != nil {
            b.l.Error("任务执行失败", logger.String("name", name), logger.Error(err))
        }
    })
}

func main() {
    // ... wire 注入
    c := cron.New(cron.WithSeconds())
    c.AddJob("@every 1m", rankingJob)
    c.Start()
    // 优雅退出
    stopCh := make(chan os.Signal, 1)
    signal.Notify(stopCh, syscall.SIGINT, syscall.SIGTERM)
    <-stopCh
    ctx := c.Stop()
    <-ctx.Done()
}

package localcache

import "sync/atomic"

// Value 泛型原子值，安全地存取整体数据
type Value[T any] struct {
    val atomic.Value // 存 *T，避免泛型零值问题
}

func (v *Value[T]) Load() (T, bool) {
    val := v.val.Load()
    if val == nil {
        var zero T
        return zero, false
    }
    return *val.(*T), true
}

func (v *Value[T]) Store(val T) {
    v.val.Store(&val)
}

type CachedRankingRepository struct {
    redis   RankCache
    local   *localcache.Value[[]domain.Article]
    exp     time.Duration // 本地缓存过期
}

func (r *CachedRankingRepository) Get(ctx context.Context) ([]domain.Article, error) {
    // 1. 优先读本地缓存
    arts, ok := r.local.Load()
    if ok {
        return arts, nil
    }
    // 2. 本地没有，读 Redis
    arts, err := r.redis.Get(ctx)
    if err != nil {
        return nil, err
    }
    // 3. 回写本地缓存
    r.local.Store(arts)
    return arts, nil
}

func (r *CachedRankingRepository) Get(ctx context.Context) ([]domain.Article, error) {
    // 优先本地缓存
    arts, ok := r.local.Load()
    if ok { return arts, nil }

    // 本地没有，尝试 Redis
    arts, err := r.redis.Get(ctx)
    if err != nil {
        // Redis 崩了：再次尝试本地缓存，不检查过期
        if arts2, ok2 := r.local.LoadForce(); ok2 {
            return arts2, nil
        }
        return nil, err
    }
    r.local.Store(arts)
    return arts, nil
}

go get github.com/gotomicro/redis-lock@latest

type DistributedRankingJob struct {
    job     job.Job
    client  *redislock.Client
    key     string
    timeout time.Duration
}

func (d *DistributedRankingJob) Run(ctx context.Context) error {
    // 尝试抢锁，过期时间 = 任务超时
    locker := d.client.Obtain(ctx, d.key, d.timeout, nil)
    if err != nil { return nil } // 别人拿到了，我跳过
    defer locker.Release(ctx)
    return d.job.Run(ctx)
}

type LongLockRankingJob struct {
    job     job.Job
    client  *redislock.Client
    key     string
    timeout time.Duration
    lock    *redislock.Lock  // 持有的锁
}

func (d *LongLockRankingJob) Run(ctx context.Context) error {
    if d.lock == nil {
        // 没拿到锁，尝试拿
        lock, err := d.client.Obtain(ctx, d.key, d.timeout, nil)
        if err != nil { return nil } // 别人有锁，跳过
        d.lock = lock
        // 拿到锁后开启自动续约
        go d.lock.Refresh(ctx, d.timeout/2, nil)
    }
    // 自己持有锁，正常执行
    return d.job.Run(ctx)
}

CREATE TABLE `task` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `name` varchar(128) NOT NULL COMMENT '任务名',
  `cron` varchar(64) NOT NULL COMMENT 'cron 表达式',
  `status` tinyint NOT NULL COMMENT '0 待执行 1 执行中',
  `next_time` bigint NOT NULL COMMENT '下次执行时间',
  `utime` bigint NOT NULL COMMENT '最后续约时间',
  `version` int NOT NULL DEFAULT 0 COMMENT '乐观锁版本号',
  PRIMARY KEY (`id`),
  KEY `idx_next_time` (`next_time`)  -- 必须索引，加快抢占查询
);

type Scheduler interface {
    Preempt(ctx context.Context) (*Task, error) // 抢占
    Release(ctx context.Context, t *Task) error // 释放
    Reset(ctx context.Context, t *Task, next time.Time) error // 重置下次时间
}

func (d *TaskDAO) Preempt(ctx context.Context, t int64) (*Task, error) {
    var task Task
    err := d.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        // 步骤 1：找到符合条件的任务：状态为待执行 + 到了执行时间
        err := tx.Where("status = ? AND next_time <= ?", 0, time.Now().UnixMilli()).
            First(&task).Error
        if err != nil { return err }
        // 步骤 2：乐观锁更新状态为执行中
        res := tx.Model(&Task{}).
            Where("id = ? AND version = ?", task.Id, task.Version).
            Updates(map[string]any{
                "status":   1,
                "version":  gorm.Expr("version + 1"),
                "utime":    time.Now().UnixMilli(),
            })
        if res.RowsAffected == 0 {
            return errors.New("抢占失败")
        }
        return nil
    })
    return &task, err
}

now := time.Now().UnixMilli()
timeout := int64(30 * 1000) // 30 秒没续约视为崩溃
db.Where(
    "(status = ? AND next_time <= ?) OR (status = ? AND ? - utime > ?)",
    0, now, 1, now, timeout,
).First(&task)

type MySQLScheduler struct {
    db       *gorm.DB
    job      job.Job
    sem      *semaphore.Weighted // 限制并发抢占数量
    timeout  time.Duration
}

func (s *MySQLScheduler) Schedule(ctx context.Context) {
    for {
        // 步骤 1：拿令牌（限制并发抢占数）
        _ = s.sem.Acquire(ctx, 1)
        go func() {
            defer s.sem.Release(1)
            // 步骤 2：抢占
            t, err := s.Preempt(ctx)
            if err != nil { return }
            // 步骤 3：启动续约
            keepAliveCtx, cancel := context.WithCancel(ctx)
            go s.KeepAlive(keepAliveCtx, t)
            // 步骤 4：执行任务
            _ = s.job.Run(ctx)
            // 步骤 5：释放
            cancel()
            _ = s.Release(ctx, t)
            // 步骤 6：重置下次执行时间
            _ = s.Reset(ctx, t, nextTime(t.Cron))
        }()
    }
}