图文帖子 推荐系统
全链路技术架构

{
  "user_id":        "10086",
  "post_id":        "p_20260205_83721",
  "event_type":     "view",           // view | like | collect | comment | share | follow | report | dislike
  "duration_ms":    8200,             // 停留时长（毫秒），仅 view 类型有值
  "image_progress": 0.75,             // 图片浏览进度 0~1，看了 4 张中的 3 张 = 0.75
  "text_expanded":  true,             // 是否点击了"展开全文"
  "comment_viewed": true,             // 是否查看了评论区
  "author_clicked": false,            // 是否点击了作者主页
  "source":         "recommend",      // 来源：recommend | follow | search | hot | profile
  "position":       7,                // 在推荐列表中的位置（第几条）
  "timestamp":      "2026-02-05T14:32:18.000Z",
  "device":         "iOS",            // iOS | Android
  "session_id":     "sess_a8f3c2e1"   // 同一次打开 App 共用一个 session
}

// 定时任务：每小时执行一次
for user in getAllActiveUsers():

    behaviors = queryClickHouse("最近 7 天的行为", user.id)
    profile = {}

    for behavior in behaviors:

        post_tags = getPostTags(behavior.post_id)

        // 不同行为给不同权重
        weight = {
            点赞: 5,
            收藏: 4,
            看完: 3,
            浏览: 1
        }[behavior.type]

        // 越近的行为权重越大
        decay = 1.0 / (1 + hours_ago * 0.02)

        for tag in post_tags:
            profile[tag] += weight * decay

    normalize(profile)  // 归一化到 0~1
    writeToRedis("user_profile:{user.id}", profile)

func MultiRecall(ctx context.Context, user *Profile) ([]Post, error) {
    ctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
    defer cancel()

    type result struct {
        posts []Post
        ch    string
    }
    ch := make(chan result, 5)

    channels := []struct {
        name string
        fn   func(context.Context, *Profile) ([]Post, error)
    }{
        {"interest", recallByInterest},  // 60% — ES Boost
        {"hot",      recallByHot},       // 15% — Redis ZSet
        {"follow",   recallByFollow},    // 15% — MySQL/Redis
        {"explore",  recallByExplore},   // 5%  — 反兴趣探索
        {"newpost",  recallByNewPost},   // 5%  — 冷启动
    }

    for _, c := range channels {
        go func(name string, fn func(context.Context, *Profile) ([]Post, error)) {
            posts, err := fn(ctx, user)
            if err != nil {
                log.Warn("recall channel %s failed: %v", name, err)
                posts = nil  // 失败不影响其他路
            }
            ch <- result{posts, name}
        }(c.name, c.fn)
    }

    var all []Post
    for i := 0; i < len(channels); i++ {
        r := <-ch
        all = append(all, r.posts...)
    }
    return mergeAndDedup(all), nil
}

-- 每 30 分钟跑一次，统计所有帖子最新互动数据
SELECT
    post_id,
    countIf(event_type IN ('like','comment','collect','share')) AS positive,
    countIf(event_type = 'view')                                AS impressions,
    countIf(event_type = 'view' AND image_progress >= 0.9)      AS finish_views,
    countIf(event_type IN ('report','dislike'))                  AS negatives,
    countIf(event_type = 'view' AND duration_ms < 1000)         AS quick_swipes
FROM behavior_events
WHERE timestamp >= now() - INTERVAL 7 DAY
GROUP BY post_id
HAVING impressions >= 10  -- 曝光太少的不更新，避免统计波动

type RankWeights struct {
    Interest  float64  // 0.35
    Quality   float64  // 0.25
    Freshness float64  // 0.20
    Hot       float64  // 0.15
    Source    float64  // 0.05
}

func RankPosts(posts []Post, user *Profile, w RankWeights) []Post {
    // 1. 批量获取各项数据
    hotScores := redis.ZMScore(ctx, "hot:global", postIDs...)

    // 2. 计算各因子的 min/max（用于归一化）
    var minES, maxES float64 = math.MaxFloat64, 0
    for _, p := range posts {
        minES = min(minES, p.ESScore)
        maxES = max(maxES, p.ESScore)
    }

    // 3. 对每篇帖子打分
    for i := range posts {
        p := &posts[i]

        interest  := normalize(p.ESScore, minES, maxES)
        quality   := p.QualityScore  // 已经是 0~1
        freshness := 1.0 / (1 + p.HoursAgo() * 0.05)
        hot       := min(hotScores[i] / hotP99, 1.0)
        source    := sourceBonus(p.RecallChannel)

        p.RankScore = interest  * w.Interest +
                      quality   * w.Quality +
                      freshness * w.Freshness +
                      hot       * w.Hot +
                      source    * w.Source
    }

    // 4. 按分数降序排列
    sort.Slice(posts, func(i, j int) bool {
        return posts[i].RankScore > posts[j].RankScore
    })
    return posts
}

const (
    maxSameCategory = 2   // 同类目最多连续 2 条
    maxSameAuthor   = 1   // 同作者不连续（最多 1 条）
)

func Scatter(ranked []Post) []Post {
    result  := make([]Post, 0, len(ranked))
    skipped := make([]Post, 0)

    for _, post := range ranked {
        if shouldSkip(result, post) {
            skipped = append(skipped, post)
        } else {
            result = append(result, post)
        }
    }

    // 把跳过的帖子插回结果中的"安全位置"
    result = insertSkipped(result, skipped)
    return result
}

func shouldSkip(result []Post, post Post) bool {
    n := len(result)

    // 规则 1: 类目不连续超过 2 条
    if n >= maxSameCategory {
        allSame := true
        for i := n - maxSameCategory; i < n; i++ {
            if result[i].Category != post.Category {
                allSame = false
                break
            }
        }
        if allSame && result[n-1].Category == post.Category {
            return true
        }
    }

    // 规则 2: 同作者不连续
    if n >= maxSameAuthor && result[n-1].AuthorID == post.AuthorID {
        return true
    }

    return false
}

func insertSkipped(result, skipped []Post) []Post {
    // 对跳过的帖子，找到第一个不违反规则的位置插入
    for _, post := range skipped {
        inserted := false
        for i := 0; i < len(result); i++ {
            if canInsertAt(result, i, post) {
                result = insertAt(result, i, post)
                inserted = true
                break
            }
        }
        if !inserted {
            result = append(result, post) // 放末尾兜底
        }
    }
    return result
}

# 创建 Bloom Filter（每用户一个）
# 预期 10000 条已读，误判率 1%
BF.RESERVE  seen:user:10086  0.01  10000

# 用户看了一篇帖子 → 标记已读
BF.ADD      seen:user:10086  post_456

# 推荐时判断帖子是否已读
BF.EXISTS   seen:user:10086  post_789    # → 0(未读) / 1(可能已读)

# 批量判断（一次检查多篇帖子）
BF.MEXISTS  seen:user:10086  post_1 post_2 post_3 post_4 post_5

func MarkSeen(ctx context.Context, userID, postID string) {
    key := "seen:" + userID
    for i := 0; i < numHashFuncs; i++ {
        pos := hashN(postID, i) % bitArraySize
        redis.SetBit(ctx, key, pos, 1)
    }
}

func IsSeen(ctx context.Context, userID, postID string) bool {
    key := "seen:" + userID
    for i := 0; i < numHashFuncs; i++ {
        pos := hashN(postID, i) % bitArraySize
        bit, _ := redis.GetBit(ctx, key, pos).Result()
        if bit == 0 {
            return false  // 有一个 0 就一定没读过
        }
    }
    return true  // 全部为 1，可能已读（有误判风险）
}

// 批量过滤：在线推荐时使用 Pipeline 减少 RTT
func FilterSeen(ctx context.Context, userID string, posts []Post) []Post {
    var unseen []Post
    for _, p := range posts {
        if !IsSeen(ctx, userID, p.ID) {
            unseen = append(unseen, p)
        }
    }
    return unseen
}

// 双 Bloom Filter 轮换
// 每 15 天轮换一次 → 实际覆盖 15~30 天的已读记录

func currentBFKey(userID string) string {
    period := time.Now().Unix() / (15 * 86400)  // 15 天一个周期
    return fmt.Sprintf("seen:%s:%d", userID, period)
}

func previousBFKey(userID string) string {
    period := time.Now().Unix() / (15 * 86400) - 1
    return fmt.Sprintf("seen:%s:%d", userID, period)
}

func MarkSeen(ctx context.Context, userID, postID string) {
    // 只写入当前周期的 BF
    redis.Do(ctx, "BF.ADD", currentBFKey(userID), postID)
}

func IsSeen(ctx context.Context, userID, postID string) bool {
    // 同时查当前和上一周期（覆盖 15~30 天）
    cur, _ := redis.Do(ctx, "BF.EXISTS", currentBFKey(userID), postID)
    if cur == 1 { return true }
    prev, _ := redis.Do(ctx, "BF.EXISTS", previousBFKey(userID), postID)
    return prev == 1
}

// 清理：设置 key 的 TTL，过期自动删除
// BF.RESERVE 后执行 EXPIRE seen:user:10086:123 2592000 (30天)