lan
8aa85ca4b2
Introduce a distributed task execution system for gRPC runners. This includes: - A `TaskBus` for cluster-mode task dispatching via Redis Pub/Sub. - A `RunnerRegistry` to track active runner instances in Redis. - Support for both standalone and cluster modes via configuration. - Refactored `RunnerHub` and `TaskManager` to use a `TaskDispatcher` interface, decoupling task submission from local execution. - Added distributed locking to `HotRankWorker` to ensure only one instance performs periodic hot rank recalculations in a multi-instance deployment. - Updated dependency injection (Wire) to support the new runner components and registry.
568 lines
14 KiB
Go
568 lines
14 KiB
Go
package service
|
||
|
||
import (
|
||
"cmp"
|
||
"context"
|
||
"errors"
|
||
"math"
|
||
"slices"
|
||
"sync"
|
||
"time"
|
||
|
||
"with_you/internal/cache"
|
||
"with_you/internal/config"
|
||
"with_you/internal/repository"
|
||
|
||
redislib "github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
type hotRankScored struct {
|
||
idx int
|
||
norm float64
|
||
}
|
||
|
||
// HotRankWorker 定时在候选池上重算热门分，写入 Redis ZSET（仅 TopN）及 top_ids（Redis+本地缓存）
|
||
// 多实例部署时通过 Redis 分布式锁保证仅主实例执行定时刷新，避免重复计算。
|
||
type HotRankWorker struct {
|
||
cfg *config.Config
|
||
postRepo repository.PostRepository
|
||
channelRepo repository.ChannelRepository
|
||
c cache.Cache
|
||
rdb *redislib.Client // 用于分布式锁（可为 nil，降级为本地运行）
|
||
mu sync.Mutex
|
||
stopOnce sync.Once
|
||
stopCh chan struct{}
|
||
doneCh chan struct{}
|
||
}
|
||
|
||
const (
|
||
// hotRankLockKey Redis 分布式锁键
|
||
hotRankLockKey = "hot_rank:leader_lock"
|
||
// hotRankLockTTL 锁持有时间（略大于刷新间隔，确保单实例持有）
|
||
hotRankLockTTL = 5 * time.Minute
|
||
// hotRankLockRetryInterval 竞争锁的重试间隔
|
||
hotRankLockRetryInterval = 30 * time.Second
|
||
)
|
||
|
||
func NewHotRankWorker(cfg *config.Config, postRepo repository.PostRepository, channelRepo repository.ChannelRepository, c cache.Cache, rdb *redislib.Client) *HotRankWorker {
|
||
return &HotRankWorker{
|
||
cfg: cfg,
|
||
postRepo: postRepo,
|
||
channelRepo: channelRepo,
|
||
c: c,
|
||
rdb: rdb,
|
||
stopCh: make(chan struct{}),
|
||
doneCh: make(chan struct{}),
|
||
}
|
||
}
|
||
|
||
// tryAcquireLock 尝试获取分布式锁，返回是否成功
|
||
func (w *HotRankWorker) tryAcquireLock(ctx context.Context) bool {
|
||
if w.rdb == nil {
|
||
return true // 无 Redis 时降级为本地运行
|
||
}
|
||
ok, err := w.rdb.SetNX(ctx, hotRankLockKey, "1", hotRankLockTTL).Result()
|
||
if err != nil {
|
||
zap.L().Warn("hot rank: failed to acquire leader lock, running locally", zap.Error(err))
|
||
return true // Redis 出错时降级为本地运行
|
||
}
|
||
return ok
|
||
}
|
||
|
||
// keepLockAlive 周期性续期分布式锁
|
||
func (w *HotRankWorker) keepLockAlive(ctx context.Context) {
|
||
if w.rdb == nil {
|
||
return
|
||
}
|
||
ticker := time.NewTicker(hotRankLockTTL / 2)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-w.stopCh:
|
||
return
|
||
case <-ticker.C:
|
||
w.rdb.Expire(ctx, hotRankLockKey, hotRankLockTTL)
|
||
}
|
||
}
|
||
}
|
||
|
||
// releaseLock 释放分布式锁
|
||
func (w *HotRankWorker) releaseLock() {
|
||
if w.rdb == nil {
|
||
return
|
||
}
|
||
w.rdb.Del(context.Background(), hotRankLockKey)
|
||
}
|
||
|
||
// Start 启动定时任务
|
||
// 多实例部署时通过 Redis 分布式锁竞争主实例角色：
|
||
// - 获得锁的实例负责执行定时刷新，并周期续期
|
||
// - 未获得锁的实例定期重试竞争（当主实例宕机时可接管）
|
||
func (w *HotRankWorker) Start(ctx context.Context) {
|
||
if w == nil || w.cfg == nil || !w.cfg.HotRank.Enabled || w.c == nil {
|
||
close(w.doneCh)
|
||
return
|
||
}
|
||
interval := time.Duration(w.cfg.HotRank.RefreshIntervalSeconds) * time.Second
|
||
if interval < 30*time.Second {
|
||
interval = 30 * time.Second
|
||
}
|
||
|
||
go func() {
|
||
defer close(w.doneCh)
|
||
|
||
isLeader := false
|
||
|
||
for {
|
||
select {
|
||
case <-w.stopCh:
|
||
if isLeader {
|
||
w.releaseLock()
|
||
}
|
||
return
|
||
default:
|
||
}
|
||
|
||
acquired := w.tryAcquireLock(ctx)
|
||
if acquired {
|
||
if !isLeader {
|
||
zap.L().Info("hot rank: acquired leader lock, becoming primary")
|
||
isLeader = true
|
||
// 启动续期协程
|
||
lockCtx, lockCancel := context.WithCancel(ctx)
|
||
go w.keepLockAlive(lockCtx)
|
||
|
||
// 主实例立即执行一次刷新
|
||
if err := w.Refresh(ctx); err != nil {
|
||
if !errors.Is(err, cache.ErrKeyNotFound) {
|
||
zap.L().Warn("hot rank initial refresh failed", zap.Error(err))
|
||
}
|
||
}
|
||
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-w.stopCh:
|
||
lockCancel()
|
||
w.releaseLock()
|
||
return
|
||
case <-ticker.C:
|
||
if err := w.Refresh(ctx); err != nil {
|
||
if !errors.Is(err, cache.ErrKeyNotFound) {
|
||
zap.L().Warn("hot rank refresh failed", zap.Error(err))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
// 不可能走到这里（isLeader==true 的分支里有无限循环）
|
||
} else {
|
||
// 未获得锁，作为备用实例等待重试
|
||
zap.L().Debug("hot rank: another instance holds leader lock, retrying later")
|
||
}
|
||
|
||
select {
|
||
case <-w.stopCh:
|
||
return
|
||
case <-time.After(hotRankLockRetryInterval):
|
||
// 重试竞争锁
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
|
||
// Stop 停止 worker
|
||
func (w *HotRankWorker) Stop() {
|
||
if w == nil {
|
||
return
|
||
}
|
||
w.stopOnce.Do(func() {
|
||
close(w.stopCh)
|
||
})
|
||
<-w.doneCh
|
||
w.releaseLock()
|
||
}
|
||
|
||
// Refresh 立即重算热门分（线程安全）
|
||
// 算法：候选 ID = 上一期 TopN（Redis）∪ 时间窗内新发帖子 ID，截断至 CandidateCap；
|
||
// 仅对候选拉取统计字段并打分，min-max 归一化；已发布置顶帖固定排在最前，其后接热门至 TopN；
|
||
// 写入 ZSET（置顶分数高于热门）及 top_ids。
|
||
func (w *HotRankWorker) Refresh(ctx context.Context) error {
|
||
if w == nil || w.c == nil {
|
||
return cache.ErrKeyNotFound
|
||
}
|
||
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
|
||
topN := w.cfg.HotRank.TopN
|
||
if topN <= 0 {
|
||
topN = 100
|
||
}
|
||
topN = min(topN, 500)
|
||
recentHours := w.cfg.HotRank.RecentWindowHours
|
||
if recentHours <= 0 {
|
||
recentHours = 168
|
||
}
|
||
recentLimit := w.cfg.HotRank.RecentFetchLimit
|
||
if recentLimit <= 0 {
|
||
recentLimit = 500
|
||
}
|
||
candidateCap := w.cfg.HotRank.CandidateCap
|
||
if candidateCap <= 0 {
|
||
candidateCap = 800
|
||
}
|
||
|
||
zsetKey := cache.HotRankZSetKey()
|
||
topIDsKey := cache.HotRankTopIDsKey()
|
||
|
||
pinned, err := w.postRepo.ListPinnedPublishedPostIDs()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
pinned = dedupeIDsPreserveOrder(pinned)
|
||
|
||
incumbent, _ := w.c.ZRevRange(ctx, zsetKey, 0, int64(topN-1))
|
||
|
||
since := time.Now().Add(-time.Duration(recentHours) * time.Hour)
|
||
recentIDs, err := w.postRepo.ListPublishedPostIDsSince(since, recentLimit)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
candidateIDs := mergeHotRankCandidates(incumbent, recentIDs, candidateCap)
|
||
if len(candidateIDs) == 0 && len(pinned) == 0 {
|
||
if err := w.c.ZReplaceSortedSet(ctx, zsetKey, nil); err != nil {
|
||
return err
|
||
}
|
||
w.c.Delete(topIDsKey)
|
||
return nil
|
||
}
|
||
|
||
var snapshots []repository.PostHotSnapshot
|
||
var ranked []hotRankScored
|
||
if len(candidateIDs) > 0 {
|
||
snapshots, err = w.postRepo.GetPostHotSnapshotsByIDs(candidateIDs)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
if len(snapshots) > 0 {
|
||
now := time.Now()
|
||
raw := make([]float64, len(snapshots))
|
||
for i := range snapshots {
|
||
p := &snapshots[i]
|
||
base := hotRankBaseScore(p.LikesCount, p.CommentsCount, p.FavoritesCount, p.SharesCount, p.ViewsCount)
|
||
tHours := now.Sub(p.CreatedAt).Hours()
|
||
if tHours < 0 {
|
||
tHours = 0
|
||
}
|
||
denom := math.Pow(tHours+2.0, 0.5)
|
||
if denom < 1e-9 {
|
||
denom = 1e-9
|
||
}
|
||
raw[i] = base / denom
|
||
}
|
||
|
||
minR := slices.Min(raw)
|
||
maxR := slices.Max(raw)
|
||
|
||
ranked = make([]hotRankScored, len(snapshots))
|
||
for i := range snapshots {
|
||
var norm float64
|
||
if maxR > minR {
|
||
norm = (raw[i] - minR) / (maxR - minR)
|
||
} else {
|
||
norm = 1
|
||
}
|
||
ranked[i] = hotRankScored{idx: i, norm: norm}
|
||
}
|
||
slices.SortFunc(ranked, func(a, b hotRankScored) int {
|
||
if a.norm != b.norm {
|
||
return cmp.Compare(b.norm, a.norm) // 降序
|
||
}
|
||
if snapshots[a.idx].CreatedAt.After(snapshots[b.idx].CreatedAt) {
|
||
return -1
|
||
} else if snapshots[a.idx].CreatedAt.Before(snapshots[b.idx].CreatedAt) {
|
||
return 1
|
||
}
|
||
return 0
|
||
})
|
||
}
|
||
|
||
pinnedSet := make(map[string]struct{}, len(pinned))
|
||
for _, id := range pinned {
|
||
pinnedSet[id] = struct{}{}
|
||
}
|
||
|
||
hotSlots := topN - len(pinned)
|
||
if hotSlots < 0 {
|
||
pinned = pinned[:topN]
|
||
pinnedSet = make(map[string]struct{}, len(pinned))
|
||
for _, id := range pinned {
|
||
pinnedSet[id] = struct{}{}
|
||
}
|
||
hotSlots = 0
|
||
}
|
||
|
||
type hotPick struct {
|
||
id string
|
||
norm float64
|
||
}
|
||
hotPicks := make([]hotPick, 0, hotSlots)
|
||
for _, r := range ranked {
|
||
id := snapshots[r.idx].ID
|
||
if _, isPin := pinnedSet[id]; isPin {
|
||
continue
|
||
}
|
||
hotPicks = append(hotPicks, hotPick{id: id, norm: r.norm})
|
||
if len(hotPicks) >= hotSlots {
|
||
break
|
||
}
|
||
}
|
||
|
||
orderedIDs := make([]string, 0, len(pinned)+len(hotPicks))
|
||
orderedIDs = append(orderedIDs, pinned...)
|
||
for _, h := range hotPicks {
|
||
orderedIDs = append(orderedIDs, h.id)
|
||
}
|
||
|
||
const pinnedScoreBase = 1000.0
|
||
zMembers := make([]redislib.Z, 0, len(orderedIDs))
|
||
for i, id := range pinned {
|
||
zMembers = append(zMembers, redislib.Z{
|
||
Score: pinnedScoreBase - float64(i)*1e-6,
|
||
Member: id,
|
||
})
|
||
}
|
||
for _, h := range hotPicks {
|
||
zMembers = append(zMembers, redislib.Z{Score: h.norm * 0.99, Member: h.id})
|
||
}
|
||
|
||
if err := w.c.ZReplaceSortedSet(ctx, zsetKey, zMembers); err != nil {
|
||
return err
|
||
}
|
||
|
||
ttl := time.Duration(w.cfg.HotRank.RefreshIntervalSeconds) * time.Second * 4
|
||
if ttl < 10*time.Minute {
|
||
ttl = 10 * time.Minute
|
||
}
|
||
w.c.Set(topIDsKey, orderedIDs, ttl)
|
||
|
||
zap.L().Info("hot rank refreshed",
|
||
zap.Int("pinned", len(pinned)),
|
||
zap.Int("candidates", len(candidateIDs)),
|
||
zap.Int("scored", len(snapshots)),
|
||
zap.Int("top_n", len(zMembers)),
|
||
)
|
||
|
||
if err := w.refreshChannelRanks(ctx, ttl); err != nil {
|
||
zap.L().Warn("hot rank channel refresh failed", zap.Error(err))
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func dedupeIDsPreserveOrder(ids []string) []string {
|
||
if len(ids) <= 1 {
|
||
return ids
|
||
}
|
||
seen := make(map[string]struct{}, len(ids))
|
||
out := make([]string, 0, len(ids))
|
||
for _, id := range ids {
|
||
if id == "" {
|
||
continue
|
||
}
|
||
if _, ok := seen[id]; ok {
|
||
continue
|
||
}
|
||
seen[id] = struct{}{}
|
||
out = append(out, id)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func mergeHotRankCandidates(incumbent, recent []string, cap int) []string {
|
||
if cap <= 0 {
|
||
return nil
|
||
}
|
||
seen := make(map[string]struct{}, cap)
|
||
out := make([]string, 0, cap)
|
||
for _, id := range incumbent {
|
||
if id == "" {
|
||
continue
|
||
}
|
||
if _, ok := seen[id]; ok {
|
||
continue
|
||
}
|
||
seen[id] = struct{}{}
|
||
out = append(out, id)
|
||
if len(out) >= cap {
|
||
return out
|
||
}
|
||
}
|
||
for _, id := range recent {
|
||
if id == "" {
|
||
continue
|
||
}
|
||
if _, ok := seen[id]; ok {
|
||
continue
|
||
}
|
||
seen[id] = struct{}{}
|
||
out = append(out, id)
|
||
if len(out) >= cap {
|
||
break
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func hotRankBaseScore(likes, comments, favorites, shares, views int) float64 {
|
||
return float64(likes)*3 +
|
||
float64(comments)*5 +
|
||
float64(favorites)*4 +
|
||
float64(shares)*6 +
|
||
float64(views)*0.15
|
||
}
|
||
|
||
func (w *HotRankWorker) refreshChannelRanks(
|
||
ctx context.Context,
|
||
ttl time.Duration,
|
||
) error {
|
||
if w.channelRepo == nil {
|
||
return nil
|
||
}
|
||
channels, err := w.channelRepo.ListActive()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(channels) == 0 {
|
||
return nil
|
||
}
|
||
|
||
topN := w.cfg.HotRank.TopN
|
||
if topN <= 0 {
|
||
topN = 100
|
||
}
|
||
topN = min(topN, 500)
|
||
|
||
for _, ch := range channels {
|
||
chCandidateIDs, cerr := w.postRepo.ListPublishedPostIDsByChannel(ch.ID, w.cfg.HotRank.RecentWindowHours, w.cfg.HotRank.CandidateCap)
|
||
if cerr != nil || len(chCandidateIDs) == 0 {
|
||
continue
|
||
}
|
||
|
||
var chRanked []hotRankScored
|
||
chSnapshots, ferr := w.postRepo.GetPostHotSnapshotsByIDs(chCandidateIDs)
|
||
if ferr != nil || len(chSnapshots) == 0 {
|
||
continue
|
||
}
|
||
|
||
now := time.Now()
|
||
raw := make([]float64, len(chSnapshots))
|
||
for i := range chSnapshots {
|
||
p := &chSnapshots[i]
|
||
base := hotRankBaseScore(p.LikesCount, p.CommentsCount, p.FavoritesCount, p.SharesCount, p.ViewsCount)
|
||
tHours := now.Sub(p.CreatedAt).Hours()
|
||
if tHours < 0 {
|
||
tHours = 0
|
||
}
|
||
denom := math.Pow(tHours+2.0, 0.5)
|
||
if denom < 1e-9 {
|
||
denom = 1e-9
|
||
}
|
||
raw[i] = base / denom
|
||
}
|
||
|
||
minR := slices.Min(raw)
|
||
maxR := slices.Max(raw)
|
||
|
||
chRanked = make([]hotRankScored, len(chSnapshots))
|
||
for i := range chSnapshots {
|
||
var norm float64
|
||
if maxR > minR {
|
||
norm = (raw[i] - minR) / (maxR - minR)
|
||
} else {
|
||
norm = 1
|
||
}
|
||
chRanked[i] = hotRankScored{idx: i, norm: norm}
|
||
}
|
||
slices.SortFunc(chRanked, func(a, b hotRankScored) int {
|
||
if a.norm != b.norm {
|
||
return cmp.Compare(b.norm, a.norm)
|
||
}
|
||
if chSnapshots[a.idx].CreatedAt.After(chSnapshots[b.idx].CreatedAt) {
|
||
return -1
|
||
} else if chSnapshots[a.idx].CreatedAt.Before(chSnapshots[b.idx].CreatedAt) {
|
||
return 1
|
||
}
|
||
return 0
|
||
})
|
||
|
||
chPinned, _ := w.postRepo.ListPinnedPublishedPostIDsByChannel(ch.ID)
|
||
chPinned = dedupeIDsPreserveOrder(chPinned)
|
||
|
||
chPinnedSet := make(map[string]struct{}, len(chPinned))
|
||
for _, id := range chPinned {
|
||
chPinnedSet[id] = struct{}{}
|
||
}
|
||
|
||
hotSlots := topN - len(chPinned)
|
||
if hotSlots < 0 {
|
||
chPinned = chPinned[:topN]
|
||
hotSlots = 0
|
||
}
|
||
|
||
type hotPick struct {
|
||
id string
|
||
norm float64
|
||
}
|
||
hotPicks := make([]hotPick, 0, hotSlots)
|
||
for _, r := range chRanked {
|
||
id := chSnapshots[r.idx].ID
|
||
if _, isPin := chPinnedSet[id]; isPin {
|
||
continue
|
||
}
|
||
hotPicks = append(hotPicks, hotPick{id: id, norm: r.norm})
|
||
if len(hotPicks) >= hotSlots {
|
||
break
|
||
}
|
||
}
|
||
|
||
chOrderedIDs := make([]string, 0, len(chPinned)+len(hotPicks))
|
||
chOrderedIDs = append(chOrderedIDs, chPinned...)
|
||
for _, h := range hotPicks {
|
||
chOrderedIDs = append(chOrderedIDs, h.id)
|
||
}
|
||
|
||
const pinnedScoreBase = 1000.0
|
||
chZMembers := make([]redislib.Z, 0, len(chOrderedIDs))
|
||
for i, id := range chPinned {
|
||
chZMembers = append(chZMembers, redislib.Z{
|
||
Score: pinnedScoreBase - float64(i)*1e-6,
|
||
Member: id,
|
||
})
|
||
}
|
||
for _, h := range hotPicks {
|
||
chZMembers = append(chZMembers, redislib.Z{Score: h.norm * 0.99, Member: h.id})
|
||
}
|
||
|
||
chZSetKey := cache.HotRankZSetChannelKey(ch.ID)
|
||
chTopIDsKey := cache.HotRankTopIDsChannelKey(ch.ID)
|
||
|
||
if err := w.c.ZReplaceSortedSet(ctx, chZSetKey, chZMembers); err != nil {
|
||
zap.L().Warn("hot rank channel ZReplaceSortedSet failed", zap.String("channel", ch.ID), zap.Error(err))
|
||
continue
|
||
}
|
||
w.c.Set(chTopIDsKey, chOrderedIDs, ttl)
|
||
}
|
||
|
||
return nil
|
||
}
|