全栈攻城狮

Appearance

Feed 流系统 / 直播互动 / 匹配推荐 ​

1. Feed 流系统 ​

1.1 Feed 流分类 ​

Feed 流系统根据内容组织方式的不同，可分为以下四类：

分类排序依据适用场景典型产品
时间线 Feed发布时间倒序社交动态、朋友圈微信朋友圈、微博时间线
社交推荐 Feed用户关系和互动信号好友动态、关注流Instagram、Facebook
热门排行 Feed热度指标综合排序发现页、热搜榜微博热搜、抖音发现
同城 FeedLBS 地理位置本地内容、附近的人抖音同城、小红书附近

时间线 Feed ​

时间线是最基础也是最容易理解的 Feed 形式，按照内容的发布时间严格倒序排列。

java
// 时间线 Feed 查询
public List<Feed> getTimelineFeed(long userId, long cursor, int pageSize) {
    // 获取用户关注列表
    Set<Long> followIds = followService.getFollowIds(userId);
    if (followIds.isEmpty()) {
        return Collections.emptyList();
    }
    // 按时间倒序拉取
    return feedMapper.selectByUserIds(followIds, cursor, pageSize,
            "ORDER BY create_time DESC");
}

社交推荐 Feed ​

在时间线基础上引入社交信号，将与用户互动多的好友内容提前。社交信号包括：点赞、评论、转发、@提及、私信互动频率等。

java
// 社交关系分计算
public double calcSocialScore(long userId, long authorId) {
    // 互动频次权重
    int interactCount = interactMapper.countInteract(userId, authorId, 7); // 近7天
    // 共同好友数
    int commonFollows = followMapper.countCommonFollow(userId, authorId);
    // 社交亲密度 = 互动频次 * 0.6 + 共同好友 * 0.4
    return interactCount * 0.6 + commonFollows * 0.4;
}

热门排行 Feed ​

热度算法综合多个维度的指标计算每条内容的热度值，按热度降序排列。

同城 Feed ​

基于用户的地理位置（GPS/WiFi/IP）获取附近用户发布的内容，结合 LBS 地理围栏技术进行内容圈定。

sql
-- 同城 Feed 查询（Geo Hash 优化）
SELECT feed_id, content, user_id, geo_hash, create_time
FROM feed
WHERE geo_hash LIKE CONCAT(#{cityGeoPrefix}, '%')
  AND create_time >= DATE_SUB(NOW(), INTERVAL 24 HOUR)
ORDER BY hot_score DESC
LIMIT 50;

1.2 Feed 推送模式对比 ​

Feed 推送主要存在三种模式：推模式（Push / Fan-out）、拉模式（Pull / Fan-out-on-load）和推拉混合模式。

维度推模式拉模式推拉混合
写放大高（粉丝越多放大越严重）低（只写一次）中（仅普通用户写扩散）
读延迟低（预计算好）高（需要合并排序）低（缓存命中率高）
实时性高中（依赖拉取频率）高
存储成本高（每粉丝一份）低（只存一份）中
大 V 友好度极差（百万粉丝写百万次）好好（大 V 走拉模式）
实现复杂度低中高

推模式（写扩散） ​

用户发布内容后，立即将内容写入所有粉丝的收件箱（Inbox）。

java
// 推模式：发布 Feed 后扇出写入粉丝收件箱
public void publishWithPush(Feed feed) {
    // 1. 保存 Feed 原内容
    feedMapper.insert(feed);

    // 2. 获取粉丝列表（仅适用于粉丝数较少的普通用户）
    List<Long> followerIds = followMapper.getFollowerIds(feed.getUserId());
    // 批量写入收件箱（异步批量写入，提升性能）
    List<InboxItem> inboxItems = followerIds.stream()
            .map(followerId -> new InboxItem(followerId, feed.getFeedId(), feed.getCreateTime()))
            .collect(Collectors.toList());
    inboxMapper.batchInsert(inboxItems);
}

拉模式（读扩散） ​

用户发布内容后只写入自己的发布列表。粉丝拉取 Feed 时，实时拉取所有关注对象的内容并进行合并排序。

java
// 拉模式：拉取时实时合并
public List<Feed> pullFeed(long userId, long cursor, int pageSize) {
    // 1. 获取关注列表
    List<Long> followIds = followMapper.getFollowIds(userId);

    // 2. 多路拉取内容（并行调用）
    List<CompletableFuture<List<Feed>>> futures = followIds.stream()
            .map(followId -> CompletableFuture.supplyAsync(
                    () -> feedMapper.selectByUserId(followId, cursor, pageSize)))
            .collect(Collectors.toList());

    // 3. 合并所有结果并按时间排序
    List<Feed> allFeeds = futures.stream()
            .map(CompletableFuture::join)
            .flatMap(Collection::stream)
            .sorted(Comparator.comparing(Feed::getCreateTime).reversed())
            .limit(pageSize)
            .collect(Collectors.toList());
    return allFeeds;
}

推拉混合模式 ​

大 V（粉丝数超过阈值）使用拉模式，普通用户使用推模式，兼具实时性和扩展性。

java
// 推拉混合模式
public void publishWithHybrid(Feed feed) {
    feedMapper.insert(feed);

    long authorId = feed.getUserId();
    int followerCount = followMapper.countFollowers(authorId);

    if (followerCount < FANOUT_THRESHOLD) {
        // 普通用户：写扩散
        fanoutService.fanoutToFollowers(feed, authorId);
    } else {
        // 大 V：只写入自己的发布列表，粉丝拉取时走拉模式
        // 同时在粉丝收件箱写入一条轻量级的"提示"用于发现
        fanoutService.fanoutToActiveFollowers(feed, authorId);
    }
}

1.3 Feed 存储设计 ​

Feed 系统的存储核心包含三张表：Feed 内容表、时间线表（收件箱）、以及用于拉取游标分页的辅助结构。

Feed 内容表 ​

存储发布的原始内容，包含文字、图片、视频等多媒体信息。

sql
CREATE TABLE feed_content (
    feed_id      BIGINT       NOT NULL PRIMARY KEY AUTO_INCREMENT,
    user_id      BIGINT       NOT NULL COMMENT '发布者 ID',
    content_type TINYINT      NOT NULL DEFAULT 1 COMMENT '1-图文 2-视频 3-转发',
    title        VARCHAR(200) DEFAULT NULL,
    content      TEXT         COMMENT '正文内容',
    media_ids    VARCHAR(1024) DEFAULT NULL COMMENT '多媒体资源 ID 列表（JSON 数组）',
    visible      TINYINT      NOT NULL DEFAULT 1 COMMENT '可见性 1-公开 2-好友 3-私密',
    create_time  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    update_time  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_user_time (user_id, create_time DESC),
    INDEX idx_create_time (create_time DESC)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

时间线表（粉丝收件箱） ​

推模式下，每个粉丝收件箱中存储 Feed ID 指针，避免冗余存储内容。

sql
CREATE TABLE feed_inbox (
    id          BIGINT   NOT NULL PRIMARY KEY AUTO_INCREMENT,
    user_id     BIGINT   NOT NULL COMMENT '收件箱所属用户（粉丝）',
    feed_id     BIGINT   NOT NULL COMMENT 'Feed 内容 ID',
    author_id   BIGINT   NOT NULL COMMENT '发布者 ID',
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Feed 发布时间',
    INDEX idx_user_time (user_id, create_time DESC),
    INDEX idx_create_time (create_time DESC)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
PARTITION BY HASH(user_id) PARTITIONS 128;

使用分区表提升查询性能，按 user_id 哈希分 128 个分区。

拉取游标分页 ​

Feed 流不适合使用传统的 OFFSET/LIMIT 分页（数据插入会导致结果漂移），采用游标分页方式。

java
// 游标分页查询
public CursorPageResult<Feed> queryInboxByCursor(long userId, long cursor, int pageSize) {
    List<FeedInbox> inboxList;
    if (cursor <= 0) {
        // 首次拉取，取最新数据
        inboxList = inboxMapper.selectByUserId(userId, pageSize);
    } else {
        // 非首次拉取，取小于游标的数据
        inboxList = inboxMapper.selectByUserIdAndCursor(userId, cursor, pageSize);
    }

    // 根据 feed_id 批量查询 Feed 内容
    List<Long> feedIds = inboxList.stream()
            .map(FeedInbox::getFeedId)
            .collect(Collectors.toList());
    List<Feed> feeds = feedContentMapper.selectBatch(feedIds);

    // 计算下一次游标
    long nextCursor = inboxList.isEmpty() ? -1 :
            inboxList.get(inboxList.size() - 1).getCreateTime().getTime();

    return new CursorPageResult<>(feeds, nextCursor, inboxList.size() >= pageSize);
}
sql
-- 游标分页 SQL
SELECT id, user_id, feed_id, author_id, create_time
FROM feed_inbox
WHERE user_id = #{userId}
  AND create_time < #{cursor}  -- cursor 为上次最后一条的 create_time 毫秒时间戳
ORDER BY create_time DESC
LIMIT #{pageSize};

1.4 Feed 热度排序 ​

热度算法用于热门排行 Feed 的内容排序，综合时间衰减、互动权重和质量分三个维度。

热度算法公式 ​

hot_score = (interact_score * quality_factor) / time_decay

其中各分量的计算方式：

  • interact_score（互动得分）：点赞数 * w1 + 评论数 * w2 + 转发数 * w3 + 收藏数 * w4
  • time_decay（时间衰减）：采用对数衰减或指数衰减，ln(e + 已过小时数) 或 e^(-lambda * t)
  • quality_factor（质量分）：基于内容原创性、图片/视频清晰度、完播率等
java
public class HotScoreCalculator {

    // 互动权重配置
    private static final double WEIGHT_LIKE = 1.0;
    private static final double WEIGHT_COMMENT = 3.0;
    private static final double WEIGHT_SHARE = 5.0;
    private static final double WEIGHT_FAVORITE = 4.0;

    // 时间衰减系数（越小衰减越快）
    private static final double DECAY_FACTOR = 0.01;

    // 牛顿冷却定律时间衰减
    public double calcTimeDecay(long createTimestamp) {
        double hoursElapsed = (System.currentTimeMillis() - createTimestamp) / 3600000.0;
        return Math.exp(DECAY_FACTOR * hoursElapsed);
    }

    // 互动得分
    public double calcInteractScore(FeedInteraction stat) {
        return stat.getLikeCount() * WEIGHT_LIKE
                + stat.getCommentCount() * WEIGHT_COMMENT
                + stat.getShareCount() * WEIGHT_SHARE
                + stat.getFavoriteCount() * WEIGHT_FAVORITE;
    }

    // 质量分（0~1）
    public double calcQualityFactor(Feed feed) {
        double score = 0.5; // 基础分
        // 原创内容加分
        if (feed.getOriginal()) score += 0.2;
        // 包含视频加分
        if (feed.hasVideo()) score += 0.1;
        // 图片清晰度评分
        score += feed.getImageQualityScore() * 0.1;
        // 内容长度评分
        double lengthScore = Math.min(1.0, feed.getContentLength() / 500.0);
        score += lengthScore * 0.1;
        return Math.min(1.0, score);
    }

    // 综合热度分
    public double calcHotScore(Feed feed, FeedInteraction stat) {
        double interactScore = calcInteractScore(stat);
        double qualityFactor = calcQualityFactor(feed);
        double timeDecay = calcTimeDecay(feed.getCreateTime().getTime());
        return (interactScore * qualityFactor) / timeDecay;
    }
}

定期重算 ​

热门 Feed 需要定期（如每 5 分钟）重算热度分，使用定时任务更新热度分到 Redis ZSET 中。

java
@Component
public class HotScoreRefreshTask {

    @Scheduled(fixedRate = 300000) // 每 5 分钟执行
    public void refreshHotScores() {
        // 获取最近 72 小时内发布的所有 Feed
        List<Feed> feeds = feedMapper.selectRecentFeeds(72);

        for (Feed feed : feeds) {
            FeedInteraction stat = interactMapper.getInteractionStats(feed.getFeedId());
            double hotScore = hotScoreCalculator.calcHotScore(feed, stat);
            // 更新到 Redis 热榜 ZSET
            redisTemplate.opsForZSet().add(
                    RedisKey.HOT_FEED_RANK,
                    String.valueOf(feed.getFeedId()),
                    hotScore);
        }

        // 只保留 TOP N
        redisTemplate.opsForZSet().removeRange(
                RedisKey.HOT_FEED_RANK, 0, -HOT_RANK_MAX_SIZE - 1);
    }
}

2. Feed 流架构 ​

2.1 发布扇出架构 ​

用户发布 Feed 后，系统通过扇出（Fan-out）机制将内容分发到粉丝收件箱。整体流程如下：

用户发布 Feed
     │
     ▼
┌─────────────────┐
│   Feed 内容写入   │  →  feed_content 表
│   DB + Cache     │  →  Redis feed:content:{feedId}
└────────┬────────┘
         │
         ▼
┌────────────────────────────────────┐
│         粉丝列表获取                │
│   follow_service.getFollowers()    │
└────────┬───────────────────────────┘
         │
         ▼
┌────────────────────────────────────┐
│       粉丝数 < 阈值 ?              │
│                                   │
│    ┌─────┴──────┐                 │
│    ▼            ▼                  │
│ 普通用户       大 V                │
│ 写扩散         写提示              │
│ (全量扇出)     (活跃粉丝扇出)       │
└────────┬───────────────────────────┘
         │
         ▼
┌─────────────────┐
│  异步批量写入     │  →  feed_inbox 表（批量 Insert）
│  消息队列削峰     │  →  MQ Topic: feed_fanout
└─────────────────┘
java
// 发布 Feed 完整流程
@Service
public class FeedPublishService {

    @Transactional
    public void publish(Feed feed) {
        // 1. 保存 Feed 内容
        feedMapper.insert(feed);

        // 2. 写入 Redis 缓存
        redisTemplate.opsForValue().set(
                RedisKey.feedContent(feed.getFeedId()),
                JSON.toJSONString(feed),
                1, TimeUnit.HOURS);

        // 3. 发送扇出消息到 MQ（异步处理）
        FanoutMessage msg = new FanoutMessage(feed.getFeedId(), feed.getUserId());
        mqTemplate.send(FeedTopic.FANOUT, msg);
    }
}

// MQ 消费者：执行扇出逻辑
@RabbitListener(queues = "feed.fanout.queue")
public void handleFanout(FanoutMessage msg) {
    int followerCount = followMapper.countFollowers(msg.getAuthorId());

    if (followerCount <= FANOUT_THRESHOLD) {
        // 普通用户：全量扇出
        fanoutAll(msg);
    } else {
        // 大 V：仅扇出给最近活跃粉丝
        fanoutActive(msg);
    }
}

private void fanoutAll(FanoutMessage msg) {
    // 分批获取粉丝 ID，每批 1000 个
    int page = 0;
    List<Long> followerIds;
    while (!(followerIds = followMapper.getFollowerPage(msg.getAuthorId(), page++, 1000)).isEmpty()) {
        List<FeedInbox> inboxItems = followerIds.stream()
                .map(fid -> new FeedInbox(fid, msg.getFeedId(), msg.getAuthorId(), new Date()))
                .collect(Collectors.toList());
        inboxMapper.batchInsert(inboxItems);
    }
}

private void fanoutActive(FanoutMessage msg) {
    // 大 V：只扇出给 7 天内登录过的活跃粉丝
    List<Long> activeFollowerIds = followMapper.getActiveFollowers(msg.getAuthorId(), 7);
    List<FeedInbox> inboxItems = activeFollowerIds.stream()
            .map(fid -> new FeedInbox(fid, msg.getFeedId(), msg.getAuthorId(), new Date()))
            .collect(Collectors.toList());
    inboxMapper.batchInsert(inboxItems);
}

2.2 拉取架构（好友 Feed 合并排序 / 缓存） ​

用户拉取 Feed 时，从收件箱读取 Feed ID 列表，批量查询 Feed 内容后合并排序输出。

用户请求 Feed
     │
     ▼
┌────────────────────┐
│  请求参数解析       │
│  userId, cursor,   │
│  pageSize          │
└────────┬───────────┘
         │
         ▼
┌──────────────────────────────────────┐
│           多级缓存查询                │
│                                      │
│  ┌──────┐   ┌──────────┐   ┌──────┐  │
│  │ L1   │   │   L2     │   │  DB  │  │
│  │本地  │ → │  Redis   │ → │      │  │
│  │Caffeine│   │          │   │      │  │
│  └──────┘   └──────────┘   └──────┘  │
└────────────────┬─────────────────────┘
                 │
                 ▼
┌────────────────────┐
│  批量查询 Feed 内容  │
│  feedContentMapper  │
│  .selectBatch()     │
└────────┬───────────┘
         │
         ▼
┌────────────────────┐
│  排序与过滤         │
│  按 create_time    │
│  降序排列           │
└────────┬───────────┘
         │
         ▼
┌────────────────────┐
│  返回 + 下一游标    │
│  (CursorPageResult)│
└────────────────────┘
java
@Service
public class FeedPullService {

    // 本地缓存（L1）
    private final Cache<Long, List<FeedInbox>> localCache = Caffeine.newBuilder()
            .maximumSize(10000)
            .expireAfterWrite(30, TimeUnit.SECONDS)
            .build();

    public CursorPageResult<Feed> pullFeed(long userId, long cursor, int pageSize) {
        // 1. 从本地缓存或 Redis 获取收件箱列表
        List<FeedInbox> inboxList = getInboxWithCache(userId, cursor, pageSize);

        if (inboxList.isEmpty()) {
            // 2. 收件箱为空，走拉模式（关注列表合并）
            inboxList = pullFromFollows(userId, cursor, pageSize);
        }

        // 3. 批量查询 Feed 内容（按 feed_id）
        List<Long> feedIds = inboxList.stream()
                .map(FeedInbox::getFeedId)
                .collect(Collectors.toList());
        Map<Long, Feed> feedMap = batchGetFeeds(feedIds);

        // 4. 组装结果
        List<Feed> feeds = inboxList.stream()
                .map(inbox -> feedMap.get(inbox.getFeedId()))
                .filter(Objects::nonNull)
                .collect(Collectors.toList());

        long nextCursor = inboxList.isEmpty() ? -1 :
                inboxList.get(inboxList.size() - 1).getCreateTime().getTime();
        boolean hasMore = inboxList.size() >= pageSize;

        return new CursorPageResult<>(feeds, nextCursor, hasMore);
    }

    // 收件箱空时，从关注列表实时拉取合并
    private List<FeedInbox> pullFromFollows(long userId, long cursor, int pageSize) {
        // 获取关注列表中的大 V（拉模式）
        List<Long> bigVIds = followMapper.getFollowBigVIds(userId);

        // 并行拉取各个大 V 的 Feed
        List<FeedInbox> merged = bigVIds.parallelStream()
                .flatMap(authorId -> {
                    List<Feed> feeds = feedMapper.selectByUserId(authorId, cursor, pageSize);
                    return feeds.stream()
                            .map(f -> new FeedInbox(userId, f.getFeedId(), authorId, f.getCreateTime()));
                })
                .sorted(Comparator.comparing(FeedInbox::getCreateTime).reversed())
                .limit(pageSize)
                .collect(Collectors.toList());

        // 合并结果写入 Redis 缓存（TTL 60s）
        if (!merged.isEmpty()) {
            String cacheKey = RedisKey.inboxCache(userId, cursor);
            redisTemplate.opsForList().rightPushAll(cacheKey, merged);
            redisTemplate.expire(cacheKey, 60, TimeUnit.SECONDS);
        }
        return merged;
    }

    // 批量获取 Feed 内容（走缓存）
    private Map<Long, Feed> batchGetFeeds(List<Long> feedIds) {
        // 批量查 Redis
        List<String> cacheKeys = feedIds.stream()
                .map(RedisKey::feedContent)
                .collect(Collectors.toList());
        List<String> cacheValues = redisTemplate.opsForValue().multiGet(cacheKeys);

        Map<Long, Feed> result = new HashMap<>();
        List<Long> missingIds = new ArrayList<>();

        for (int i = 0; i < feedIds.size(); i++) {
            String val = cacheValues.get(i);
            if (val != null) {
                result.put(feedIds.get(i), JSON.parseObject(val, Feed.class));
            } else {
                missingIds.add(feedIds.get(i));
            }
        }

        // 缓存缺失的查 DB
        if (!missingIds.isEmpty()) {
            List<Feed> dbFeeds = feedMapper.selectBatch(missingIds);
            for (Feed f : dbFeeds) {
                result.put(f.getFeedId(), f);
                // 回填缓存
                redisTemplate.opsForValue().set(
                        RedisKey.feedContent(f.getFeedId()),
                        JSON.toJSONString(f), 1, TimeUnit.HOURS);
            }
        }
        return result;
    }
}

2.3 Feed 快慢路径 ​

大 V 和普通用户采取不同的 Feed 分发路径，在实时性和系统开销之间取得平衡。

快路径（普通用户写扩散） ​

发布 → MQ → 扇出写入所有粉丝收件箱 → 粉丝拉取直接读收件箱（命中率高，延迟低）
  • 延迟低：收件箱数据已预计算
  • 写放大：扇出写入量大但可控（普通用户粉丝有限）
  • 缓存命中率高：收件箱几乎总是包含最新内容

慢路径（大 V 拉模式） ​

发布 → 只写大 V 自己的发布列表 → 粉丝拉取时实时合并大 V Feed（缓存 60s）
  • 延迟略高：粉丝端需要实时合并
  • 写放大：几乎为零（只写一次）
  • 适用条件：粉丝数超过阈值（如 10 万）的用户
java
// 判断 Feed 走快路径还是慢路径
public FeedPath decidePath(long authorId) {
    int followerCount = followMapper.countFollowers(authorId);
    if (followerCount <= FANOUT_THRESHOLD) {
        return FeedPath.FAST_PATH;   // 快路径 - 写扩散
    } else {
        return FeedPath.SLOW_PATH;   // 慢路径 - 拉模式
    }
}

enum FeedPath {
    FAST_PATH,  // 写扩散
    SLOW_PATH   // 拉模式
}

混合路径的收件箱补充 ​

大 V 发布后，系统仍然会向"最近活跃粉丝"进行写扩散，以保证这部分粉丝能获得低延迟体验。其余非活跃粉丝在登录时通过拉模式获取。

大 V 发布 Feed
    │
    ├─→ 活跃粉丝（7天内登录）：写扩散（快路径） → 收件箱
    └─→ 非活跃粉丝：不写入，登录时拉取合并（慢路径） → 实时合并

3. 直播互动 ​

3.1 直播间架构 ​

直播系统涉及推流、转码、分发、拉流四个核心环节。

整体架构 ​

主播推流
    │
    ▼
┌────────────────────────────────────────┐
│           推流接入层                    │
│  RTMP / SRT / WebRTC                   │
│  支持弱网自适应、丢包重传               │
└────────────────┬───────────────────────┘
                 │
                 ▼
┌────────────────────────────────────────┐
│           转码集群                     │
│  ┌──────────┐  ┌──────────┐           │
│  │ 视频转码  │  │ 音频转码  │           │
│  │ H.264    │  │ AAC      │           │
│  │ H.265/HEVC│  │ OPUS     │           │
│  │ AV1      │  │          │           │
│  └──────────┘  └──────────┘           │
│  输出多码率：流畅(480p) 高清(720p)    │
│  超清(1080p) 4K(2160p)                │
└────────────────┬───────────────────────┘
                 │
                 ▼
┌────────────────────────────────────────┐
│           CDN 分发网络                 │
│  边缘节点缓存 + 回源                   │
└────────────────┬───────────────────────┘
                 │
                 ▼
┌────────────────────────────────────────┐
│           拉流播放                     │
│  FLV (低延迟 HLS) / HLS (兼容性好)    │
│  WebRTC (超低延迟)                      │
└────────────────────────────────────────┘

推流协议对比 ​

协议延迟优点缺点
RTMP2~5s成熟稳定，生态好基于 TCP，弱网表现一般
SRT0.5~2s弱网抗丢包强，支持 AES 加密较新，部分 CDN 不支持
WebRTC<500ms超低延迟，P2P 能力大规模分发成本高
HLS5~30s兼容性好，支持 HTML5延迟较高

推流接入示例 ​

java
// 创建直播房间并获取推流地址
@Service
public class LiveRoomService {

    public LiveRoom createRoom(Long anchorId, String title, LiveConfig config) {
        // 1. 生成推流地址（带鉴权 token）
        String streamKey = UUID.randomUUID().toString().replace("-", "");
        String pushUrl = String.format("rtmp://push.live.example.com/live/%s?token=%s&expire=%d",
                streamKey, generateToken(streamKey), System.currentTimeMillis() + 7200000);

        // 2. 生成播放地址（多协议）
        String playFlv = String.format("https://play.live.example.com/live/%s.flv", streamKey);
        String playHls = String.format("https://play.live.example.com/live/%s.m3u8", streamKey);

        // 3. 创建房间记录
        LiveRoom room = new LiveRoom();
        room.setAnchorId(anchorId);
        room.setTitle(title);
        room.setStreamKey(streamKey);
        room.setPushUrl(pushUrl);
        room.setPlayFlv(playFlv);
        room.setPlayHls(playHls);
        room.setStatus(LiveStatus.NOT_STARTED);
        room.setCreateTime(new Date());
        liveRoomMapper.insert(room);

        // 4. 初始化房间状态到 Redis
        redisTemplate.opsForValue().set(
                RedisKey.liveRoomInfo(room.getRoomId()),
                JSON.toJSONString(room), 24, TimeUnit.HOURS);
        return room;
    }

    private String generateToken(String streamKey) {
        String data = streamKey + "secret_key_live_2024" + (System.currentTimeMillis() / 10000);
        return DigestUtils.md5DigestAsHex(data.getBytes(StandardCharsets.UTF_8));
    }
}

CDN 转码配置 ​

yaml
# 转码模板配置
transcode:
  templates:
    - name: fluent           # 流畅
      video_codec: h264
      width: 640
      height: 360
      bitrate: 400k
      fps: 24
    - name: hd               # 高清
      video_codec: h264
      width: 1280
      height: 720
      bitrate: 1500k
      fps: 30
    - name: ultra_hd          # 超清
      video_codec: h265
      width: 1920
      height: 1080
      bitrate: 3000k
      fps: 30

3.2 直播间实时互动 ​

直播间的实时互动主要包括弹幕、礼物打赏和连麦 PK 三大场景。

弹幕系统（WebSocket） ​

弹幕使用 WebSocket 长连接实现实时消息推送，架构如下：

WebSocket 连接管理器
    │
    ▼
┌────────────────────────────────────────┐
│           WebSocket Server             │
│  Netty / Spring WebSocket              │
│  连接数: 单机 10 万+                    │
│  心跳间隔: 30s                          │
└────────────────┬───────────────────────┘
                 │
                 ▼
┌────────────────────────────────────────┐
│          消息分发层                    │
│  弹幕消息 → MQ Topic: danmu            │
│  礼物消息 → MQ Topic: gift             │
│  系统消息 → 直接广播                   │
└────────────────┬───────────────────────┘
                 │
                 ▼
┌────────────────────────────────────────┐
│         房间内广播                     │
│  按 roomId 维度广播                   │
│  合并写 + 频率限制                     │
└────────────────────────────────────────┘
java
// WebSocket 弹幕处理器
@Component
public class DanmuWebSocketHandler extends TextWebSocketHandler {

    // roomId -> Set<SessionId>
    private final Map<Long, Set<String>> roomSessions = new ConcurrentHashMap<>();
    // SessionId -> WebSocketSession
    private final Map<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        String sessionId = session.getId();
        sessionMap.put(sessionId, session);

        // 从握手参数中获取 roomId
        Long roomId = extractRoomId(session);
        roomSessions.computeIfAbsent(roomId, k -> ConcurrentHashMap.newKeySet()).add(sessionId);

        // 加入房间通知
        broadcastToRoom(roomId, SystemMessage.userEnter(sessionId));
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) {
        DanmuMessage danmu = JSON.parseObject(message.getPayload(), DanmuMessage.class);

        // 1. 频率限制（同一用户每秒最多 3 条弹幕）
        String rateKey = RedisKey.danmuRateLimit(danmu.getUserId());
        Long count = redisTemplate.opsForValue().increment(rateKey);
        if (count != null && count == 1) {
            redisTemplate.expire(rateKey, 1, TimeUnit.SECONDS);
        }
        if (count != null && count > 3) {
            sendToUser(session, SystemMessage.rateLimit("弹幕发送过于频繁"));
            return;
        }

        // 2. 内容审核（异步）
        auditService.auditText(danmu.getContent());

        // 3. 发送到 MQ 异步处理
        mqTemplate.convertAndSend(LiveTopic.DANMU, danmu);

        // 4. 广播给房间内所有用户
        broadcastToRoom(danmu.getRoomId(), danmu);
    }

    // 房间内广播
    private void broadcastToRoom(Long roomId, Object message) {
        Set<String> sessionIds = roomSessions.get(roomId);
        if (sessionIds == null || sessionIds.isEmpty()) return;

        String payload = JSON.toJSONString(message);
        TextMessage textMsg = new TextMessage(payload);

        sessionIds.forEach(sessionId -> {
            WebSocketSession session = sessionMap.get(sessionId);
            if (session != null && session.isOpen()) {
                try {
                    session.sendMessage(textMsg);
                } catch (IOException e) {
                    log.error("Send danmu failed, sessionId={}", sessionId, e);
                }
            }
        });
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        String sessionId = session.getId();
        Long roomId = extractRoomId(session);
        Optional.ofNullable(roomSessions.get(roomId))
                .ifPresent(s -> s.remove(sessionId));
        sessionMap.remove(sessionId);
    }
}

礼物打赏 ​

礼物打赏涉及实时送礼动画展示和扣款流程，要求高可靠和低延迟。

java
// 礼物打赏服务
@Service
public class GiftService {

    @Transactional
    public GiftResult sendGift(GiftSendRequest request) {
        // 1. 校验礼物信息
        Gift gift = giftMapper.selectById(request.getGiftId());
        if (gift == null) {
            throw new BusinessException("礼物不存在");
        }

        // 2. 校验余额
        UserWallet wallet = walletMapper.selectByUserIdForUpdate(request.getUserId());
        long totalAmount = gift.getPrice() * request.getQuantity();
        if (wallet.getBalance() < totalAmount) {
            throw new BusinessException("余额不足");
        }

        // 3. 扣减余额
        walletMapper.deductBalance(request.getUserId(), totalAmount);

        // 4. 记录流水
        GiftFlow flow = new GiftFlow();
        flow.setUserId(request.getUserId());
        flow.setAnchorId(request.getAnchorId());
        flow.setGiftId(request.getGiftId());
        flow.setQuantity(request.getQuantity());
        flow.setAmount(totalAmount);
        flow.setRoomId(request.getRoomId());
        flow.setCreateTime(new Date());
        giftFlowMapper.insert(flow);

        // 5. 实时推送礼物特效消息（WebSocket）
        GiftEffectMessage effectMsg = new GiftEffectMessage(
                request.getUserId(), request.getAnchorId(),
                gift, request.getQuantity());
        mqTemplate.convertAndSend(LiveTopic.GIFT, effectMsg);

        // 6. 主播收入增加（异步）
        mqTemplate.convertAndSend(LiveTopic.ANCHOR_INCOME,
                new AnchorIncomeMessage(request.getAnchorId(), totalAmount));

        return new GiftResult(flow.getFlowId(), true, "打赏成功");
    }
}

连麦 PK ​

连麦 PK 通过 WebRTC 实现主播间的实时音视频通信。

主播 A                   信令服务器                   主播 B
   │                        │                          │
   │────── 发起 PK ────────→│                          │
   │                        │────── PK 邀请 ──────────→│
   │                        │                          │
   │                        │←────── 接受 PK ──────────│
   │←───── PK 开始 ────────│                          │
   │                        │                          │
   │←────── WebRTC Offer ───│───────── WebRTC Offer ──→│
   │────── WebRTC Answer ──→│←─────── WebRTC Answer ───│
   │                        │                          │
   │←────────────────── ICE Candidate 交换 ────────────→│
   │                        │                          │
   │←────────────────── P2P 音视频流 ──────────────────→│
   │                        │                          │
   │──── PK 结束(30s倒计时)─→│                          │
   │                        │──── PK 结果通知 ────────→│
java
// 连麦 PK 信令服务
@Service
public class PKService {

    // 发起 PK 请求
    public PKResult startPK(Long roomIdA, Long roomIdB) {
        // 1. 校验两个直播间都在直播中
        LiveRoom roomA = liveRoomMapper.selectById(roomIdA);
        LiveRoom roomB = liveRoomMapper.selectById(roomIdB);
        if (roomA.getStatus() != LiveStatus.LIVE ||
            roomB.getStatus() != LiveStatus.LIVE) {
            throw new BusinessException("直播间不在直播状态");
        }

        // 2. 创建 PK 记录
        PKMatch pk = new PKMatch();
        pk.setRoomIdA(roomIdA);
        pk.setRoomIdB(roomIdB);
        pk.setAnchorIdA(roomA.getAnchorId());
        pk.setAnchorIdB(roomB.getAnchorId());
        pk.setStartTime(new Date());
        pk.setDurationSeconds(300); // PK 时长 5 分钟
        pk.setStatus(PKStatus.ONGOING);
        pkMatchMapper.insert(pk);

        // 3. 初始化 PK 计数 Redis
        String pkScoreKey = RedisKey.pkScore(pk.getPkId());
        redisTemplate.opsForValue().set(pkScoreKey + ":A", "0");
        redisTemplate.opsForValue().set(pkScoreKey + ":B", "0");
        redisTemplate.expire(pkScoreKey + ":A", 10, TimeUnit.MINUTES);
        redisTemplate.expire(pkScoreKey + ":B", 10, TimeUnit.MINUTES);

        // 4. 通过 WebSocket 通知双方开始 PK
        wsService.sendToRoom(roomIdA, PKMessage.pkStart(pk.getPkId(), roomIdB));
        wsService.sendToRoom(roomIdB, PKMessage.pkStart(pk.getPkId(), roomIdA));

        return new PKResult(pk.getPkId(), true, "PK 开始");
    }

    // PK 礼物计分（PK 期间礼物分值翻倍）
    public void addPKScore(Long pkId, String side, long giftAmount) {
        String key = RedisKey.pkScore(pkId) + ":" + side;
        redisTemplate.opsForValue().increment(key, giftAmount * 2); // PK 翻倍
    }

    // PK 结束结算
    @Scheduled(fixedDelay = 1000)
    public void checkPKTimeout() {
        List<PKMatch> ongoingList = pkMatchMapper.selectOngoing();
        for (PKMatch pk : ongoingList) {
            if (pk.getStartTime().getTime() + pk.getDurationSeconds() * 1000L < System.currentTimeMillis()) {
                finishPK(pk);
            }
        }
    }

    private void finishPK(PKMatch pk) {
        long scoreA = Optional.ofNullable(
                redisTemplate.opsForValue().get(RedisKey.pkScore(pk.getPkId()) + ":A"))
                .map(s -> Long.parseLong(s.toString())).orElse(0L);
        long scoreB = Optional.ofNullable(
                redisTemplate.opsForValue().get(RedisKey.pkScore(pk.getPkId()) + ":B"))
                .map(s -> Long.parseLong(s.toString())).orElse(0L);

        pk.setScoreA(scoreA);
        pk.setScoreB(scoreB);
        pk.setWinner(scoreA >= scoreB ? pk.getAnchorIdA() : pk.getAnchorIdB());
        pk.setStatus(PKStatus.FINISHED);
        pk.setEndTime(new Date());
        pkMatchMapper.updateById(pk);

        // 通知双方 PK 结果
        wsService.sendToRoom(pk.getRoomIdA(), PKMessage.pkResult(pk));
        wsService.sendToRoom(pk.getRoomIdB(), PKMessage.pkResult(pk));
    }
}

3.3 礼物系统 ​

礼物系统包含礼物配置、礼物特效、充值体系和流水记录四个核心模块。

礼物表设计 ​

sql
-- 礼物配置表
CREATE TABLE gift (
    gift_id     INT          NOT NULL PRIMARY KEY AUTO_INCREMENT,
    name        VARCHAR(50)  NOT NULL COMMENT '礼物名称',
    price       BIGINT       NOT NULL COMMENT '价格（虚拟币）',
    category    TINYINT      NOT NULL DEFAULT 1 COMMENT '1-普通 2-特效 3-活动限定',
    animation_url VARCHAR(255) DEFAULT NULL COMMENT '礼物动效资源地址',
    icon_url    VARCHAR(255) NOT NULL COMMENT '礼物图标',
    sort_order  INT          NOT NULL DEFAULT 0 COMMENT '排序权重',
    status      TINYINT      NOT NULL DEFAULT 1 COMMENT '1-上架 0-下架',
    create_time DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
sql
-- 用户钱包表
CREATE TABLE user_wallet (
    wallet_id   BIGINT      NOT NULL PRIMARY KEY AUTO_INCREMENT,
    user_id     BIGINT      NOT NULL UNIQUE,
    balance     BIGINT      NOT NULL DEFAULT 0 COMMENT '虚拟币余额',
    total_recharge BIGINT   NOT NULL DEFAULT 0 COMMENT '累计充值',
    total_spend BIGINT      NOT NULL DEFAULT 0 COMMENT '累计消费',
    version     INT         NOT NULL DEFAULT 0 COMMENT '乐观锁版本',
    create_time DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP,
    update_time DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
sql
-- 礼物流水表
CREATE TABLE gift_flow (
    flow_id     BIGINT      NOT NULL PRIMARY KEY AUTO_INCREMENT,
    user_id     BIGINT      NOT NULL COMMENT '送礼用户',
    anchor_id   BIGINT      NOT NULL COMMENT '主播 ID',
    gift_id     INT         NOT NULL COMMENT '礼物 ID',
    quantity    INT         NOT NULL DEFAULT 1,
    amount      BIGINT      NOT NULL COMMENT '总金额',
    room_id     BIGINT      NOT NULL COMMENT '直播间 ID',
    pk_id       BIGINT      DEFAULT NULL COMMENT '关联 PK ID',
    create_time DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_anchor_time (anchor_id, create_time DESC),
    INDEX idx_user_time (user_id, create_time DESC),
    INDEX idx_room_time (room_id, create_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
sql
-- 充值订单表
CREATE TABLE recharge_order (
    order_id     BIGINT       NOT NULL PRIMARY KEY AUTO_INCREMENT,
    user_id      BIGINT       NOT NULL,
    amount       BIGINT       NOT NULL COMMENT '充值金额（分）',
    coin_amount  BIGINT       NOT NULL COMMENT '获得虚拟币数',
    pay_type     TINYINT      NOT NULL COMMENT '1-微信 2-支付宝 3-苹果内购',
    status       TINYINT      NOT NULL DEFAULT 0 COMMENT '0-待支付 1-支付成功 2-已退款',
    pay_time     DATETIME     DEFAULT NULL,
    create_time  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_user_id (user_id),
    INDEX idx_status (status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

充值流程 ​

java
@Service
public class RechargeService {

    @Transactional
    public RechargeResult createOrder(Long userId, Long amount, Integer payType) {
        // 1. 计算赠送比例（充得越多赠送越多）
        long coinAmount = amount * 10; // 基础 1:10
        if (amount >= 64800) { // 648 元档位
            coinAmount += amount * 5;  // 额外赠送 50%
        } else if (amount >= 19800) {
            coinAmount += amount * 3;  // 额外赠送 30%
        } else if (amount >= 6000) {
            coinAmount += amount * 2;  // 额外赠送 20%
        }

        // 2. 创建订单
        RechargeOrder order = new RechargeOrder();
        order.setUserId(userId);
        order.setAmount(amount);
        order.setCoinAmount(coinAmount);
        order.setPayType(payType);
        order.setStatus(0);
        order.setCreateTime(new Date());
        rechargeOrderMapper.insert(order);

        // 3. 调用支付网关获取支付链接
        String payUrl = payGateway.createPayOrder(
                order.getOrderId(), amount, payType,
                "充值" + coinAmount + "虚拟币");

        return new RechargeResult(order.getOrderId(), payUrl, coinAmount);
    }

    // 支付回调处理
    @Transactional
    public void handlePayCallback(PayCallback callback) {
        RechargeOrder order = rechargeOrderMapper.selectByIdForUpdate(callback.getOrderId());
        if (order == null || order.getStatus() != 0) {
            return;
        }

        // 更新订单状态
        order.setStatus(1);
        order.setPayTime(new Date());
        rechargeOrderMapper.updateById(order);

        // 增加钱包余额
        walletMapper.addBalance(order.getUserId(), order.getCoinAmount());
    }
}

3.4 直播间状态管理 ​

直播间存在三种核心状态以及对应的状态转换。

状态定义 ​

状态编码说明
未开播NOT_STARTED房间已创建，主播尚未推流
直播中LIVE主播正在推流，观众可观看
回放PLAYBACK直播结束，生成回放文件

状态转换 ​

NOT_STARTED ──(推流回调)──→ LIVE ──(断流/主动结束)──→ PLAYBACK
                                    └──(断流超时)──→ NOT_STARTED（未正式开播）
java
// 直播间状态机
@Component
public class LiveRoomStateMachine {

    // 推流开始回调（由流媒体服务器回调）
    public void onStreamPublished(String streamKey) {
        LiveRoom room = liveRoomMapper.selectByStreamKey(streamKey);
        if (room == null) return;

        if (room.getStatus() == LiveStatus.NOT_STARTED) {
            room.setStatus(LiveStatus.LIVE);
            room.setLiveStartTime(new Date());
            liveRoomMapper.updateStatus(room);

            // 通知所有关注者主播开播了
            mqTemplate.convertAndSend(FeedTopic.LIVE_NOTIFY,
                    new LiveNotifyMessage(room.getAnchorId(), room.getRoomId()));
        }
    }

    // 推流结束回调（由流媒体服务器回调）
    public void onStreamUnpublished(String streamKey) {
        LiveRoom room = liveRoomMapper.selectByStreamKey(streamKey);
        if (room == null || room.getStatus() != LiveStatus.LIVE) return;

        // 判断是否正式开播（推流时长超过 30 秒才生成回放）
        long liveDuration = System.currentTimeMillis() - room.getLiveStartTime().getTime();
        if (liveDuration > 30000) {
            // 生成回放
            room.setStatus(LiveStatus.PLAYBACK);
            room.setLiveEndTime(new Date());
            // 异步生成回放文件
            mqTemplate.convertAndSend(LiveTopic.GENERATE_REPLAY,
                    new ReplayTask(room.getRoomId(), streamKey));
        } else {
            // 未正式开播，回到未开播状态
            room.setStatus(LiveStatus.NOT_STARTED);
        }
        liveRoomMapper.updateStatus(room);
    }

    // 回放就绪回调
    public void onReplayReady(Long roomId, String replayUrl) {
        LiveRoom room = liveRoomMapper.selectById(roomId);
        room.setReplayUrl(replayUrl);
        liveRoomMapper.updateReplayUrl(room);
    }
}

状态查询缓存 ​

直播间状态需要高并发读取，使用 Redis 缓存状态信息。

java
public LiveRoomStatus getRoomStatus(Long roomId) {
    // 先从 Redis 查
    String cacheKey = RedisKey.liveRoomStatus(roomId);
    String cached = (String) redisTemplate.opsForValue().get(cacheKey);
    if (cached != null) {
        return LiveRoomStatus.valueOf(cached);
    }

    // Redis 查不到，查 DB 并回填
    LiveRoom room = liveRoomMapper.selectById(roomId);
    if (room == null) return null;

    LiveRoomStatus status = room.getStatus();
    // 直播中的数据设置较短 TTL（30s），确保状态及时更新
    int ttl = status == LiveStatus.LIVE ? 30 : 300;
    redisTemplate.opsForValue().set(cacheKey, status.name(), ttl, TimeUnit.SECONDS);
    return status;
}

4. 匹配推荐 ​

4.1 匹配算法 ​

匹配推荐系统采用多种算法协同工作，以下为主要算法：

协同过滤 ​

协同过滤分为基于用户（UserCF）和基于物品（ItemCF）两种。

基于用户的协同过滤（UserCF）

核心思路：找到与目标用户兴趣相似的用户群，推荐这些用户喜欢的内容。

java
@Service
public class UserCFRecommender {

    // 计算用户相似度矩阵（Pearson 相关系数）
    public Map<Long, Double> calcSimilarUsers(Long targetUserId, int topN) {
        // 1. 获取目标用户的互动向量（喜爱的内容 ID 集合）
        Set<Long> targetItems = interactMapper.getLikedItemIds(targetUserId);

        // 2. 找到有过共同互动的用户
        List<Long> candidateUsers = interactMapper.findUsersWithCommonItems(targetItems);

        // 3. 计算每个候选用户的相似度
        Map<Long, Double> similarityMap = new HashMap<>();
        for (Long candidateId : candidateUsers) {
            if (candidateId.equals(targetUserId)) continue;
            Set<Long> candidateItems = interactMapper.getLikedItemIds(candidateId);

            double similarity = calcPearsonSimilarity(targetItems, candidateItems);
            if (similarity > 0.1) { // 过滤低相似度
                similarityMap.put(candidateId, similarity);
            }
        }

        // 4. 取 TopN
        return similarityMap.entrySet().stream()
                .sorted(Map.Entry.<Long, Double>comparingByValue().reversed())
                .limit(topN)
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

    // Pearson 相关系数
    private double calcPearsonSimilarity(Set<Long> setA, Set<Long> setB) {
        Set<Long> intersection = new HashSet<>(setA);
        intersection.retainAll(setB);

        Set<Long> union = new HashSet<>(setA);
        union.addAll(setB);

        if (union.isEmpty()) return 0;
        return (double) intersection.size() / union.size(); // Jaccard 相似度
    }

    // 基于相似用户推荐
    public List<Long> recommendByUsers(Long targetUserId, int topN) {
        Map<Long, Double> similarUsers = calcSimilarUsers(targetUserId, topN * 2);
        Set<Long> targetItems = interactMapper.getLikedItemIds(targetUserId);

        // 候选物品加权评分
        Map<Long, Double> itemScores = new HashMap<>();
        for (Map.Entry<Long, Double> userEntry : similarUsers.entrySet()) {
            Long similarUserId = userEntry.getKey();
            double similarity = userEntry.getValue();

            Set<Long> candidateItems = interactMapper.getLikedItemIds(similarUserId);
            for (Long itemId : candidateItems) {
                if (targetItems.contains(itemId)) continue; // 排除已互动过的
                itemScores.merge(itemId, similarity, Double::sum);
            }
        }

        return itemScores.entrySet().stream()
                .sorted(Map.Entry.<Long, Double>comparingByValue().reversed())
                .limit(topN)
                .map(Map.Entry::getKey)
                .collect(Collectors.toList());
    }
}

基于物品的协同过滤（ItemCF）

核心思路：计算物品之间的相似度（通常基于用户共同行为），推荐与用户喜欢物品相似的内容。

java
@Service
public class ItemCFRecommender {

    // 计算物品相似度矩阵（离线预计算，存入 Redis）
    @Scheduled(cron = "0 0 3 * * ?") // 每天凌晨 3 点更新
    public void preComputeItemSimilarity() {
        // 1. 获取所有用户互动记录（用户 -> 物品列表）
        List<UserItemInteraction> interactions = interactMapper.getAllInteractions();

        // 2. 构建物品共现矩阵
        Map<Long, Map<Long, Integer>> cooccurrence = new HashMap<>();
        Map<Long, Integer> itemUserCount = new HashMap<>();

        for (UserItemInteraction interaction : interactions) {
            List<Long> items = interaction.getItemIds();
            // 同用户下的物品构成共现关系
            for (int i = 0; i < items.size(); i++) {
                Long itemA = items.get(i);
                itemUserCount.merge(itemA, 1, Integer::sum);
                for (int j = i + 1; j < items.size(); j++) {
                    Long itemB = items.get(j);
                    cooccurrence.computeIfAbsent(itemA, k -> new HashMap<>())
                            .merge(itemB, 1, Integer::sum);
                    cooccurrence.computeIfAbsent(itemB, k -> new HashMap<>())
                            .merge(itemA, 1, Integer::sum);
                }
            }
        }

        // 3. 计算余弦相似度并存入 Redis
        int totalUsers = interactions.size();
        for (Map.Entry<Long, Map<Long, Integer>> entry : cooccurrence.entrySet()) {
            Long itemA = entry.getKey();
            double normA = Math.sqrt(itemUserCount.getOrDefault(itemA, 0));

            Map<Long, Double> similarityMap = new HashMap<>();
            for (Map.Entry<Long, Integer> coEntry : entry.getValue().entrySet()) {
                Long itemB = coEntry.getKey();
                int coCount = coEntry.getValue();
                double normB = Math.sqrt(itemUserCount.getOrDefault(itemB, 0));
                // 余弦相似度 = 共现数 / sqrt(|A| * |B|) ，并做降权（热门物品惩罚）
                double similarity = coCount / (normA * normB + 1e-10);
                // 热门物品惩罚（越热门降权越多）
                double popularityPenalty = 1.0 / Math.log(1 + itemUserCount.getOrDefault(itemB, 0));
                similarityMap.put(itemB, similarity * popularityPenalty);
            }

            // 只保留 Top 100 最相似物品
            List<Map.Entry<Long, Double>> topSimilar = similarityMap.entrySet().stream()
                    .sorted(Map.Entry.<Long, Double>comparingByValue().reversed())
                    .limit(100)
                    .collect(Collectors.toList());

            String redisKey = RedisKey.itemSimilarity(itemA);
            Map<String, Double> redisMap = new HashMap<>();
            for (Map.Entry<Long, Double> sEntry : topSimilar) {
                redisMap.put(String.valueOf(sEntry.getKey()), sEntry.getValue());
            }
            redisTemplate.opsForZSet().add(redisKey, redisMap);
            redisTemplate.expire(redisKey, 24, TimeUnit.HOURS);
        }
    }

    // 根据物品相似度推荐
    public List<Long> recommendByItems(Long userId, Long likedItemId, int topN) {
        Set<Long> interactedItems = interactMapper.getLikedItemIds(userId);

        String redisKey = RedisKey.itemSimilarity(likedItemId);
        Set<ZSetOperations.TypedTuple<String>> similarItems =
                redisTemplate.opsForZSet().reverseRangeWithScores(redisKey, 0, topN * 2);

        return similarItems.stream()
                .map(tuple -> Long.parseLong(tuple.getValue()))
                .filter(itemId -> !interactedItems.contains(itemId))
                .limit(topN)
                .collect(Collectors.toList());
    }
}

标签匹配 ​

基于用户和内容的标签进行匹配。用户在注册或使用过程中标记兴趣标签，内容发布时标注内容标签。

java
// 标签匹配推荐
public List<Long> recommendByTags(Long userId, int topN) {
    // 1. 获取用户兴趣标签及权重
    List<UserTag> userTags = tagMapper.selectUserTags(userId);

    // 2. 根据标签匹配未读过的内容
    Map<Long, Double> itemScores = new HashMap<>();
    for (UserTag userTag : userTags) {
        List<Feed> taggedFeeds = feedMapper.selectByTag(userTag.getTagId(), topN);
        for (Feed feed : taggedFeeds) {
            itemScores.merge(feed.getFeedId(), userTag.getWeight(), Double::sum);
        }
    }

    // 3. 按匹配度排序
    return itemScores.entrySet().stream()
            .sorted(Map.Entry.<Long, Double>comparingByValue().reversed())
            .limit(topN)
            .map(Map.Entry::getKey)
            .collect(Collectors.toList());
}

LBS 地理位置匹配 ​

基于用户当前 GPS 坐标，推荐附近的内容或用户。

java
// LBS 匹配推荐（基于 GeoHash）
public List<Feed> recommendByLBS(double latitude, double longitude, int radiusKm, int topN) {
    // 1. 根据范围计算 GeoHash 前缀
    String geoHash = GeoHashUtils.encode(latitude, longitude, 6); // 约 ±0.6km
    String geoPrefix = geoHash.substring(0, 5); // 约 ±2.5km

    // 2. 扩大/缩小前缀匹配范围
    int precision = 6;
    if (radiusKm > 10) precision = 4; // 约 ±20km
    else if (radiusKm > 5) precision = 5; // 约 ±2.5km

    // 3. 查询附近 Feed
    List<Feed> nearbyFeeds = feedMapper.selectByGeoPrefix(
            geoHash.substring(0, precision), topN * 3);

    // 4. 精确距离过滤和排序
    return nearbyFeeds.stream()
            .filter(feed -> {
                double distance = GeoHashUtils.distance(
                        latitude, longitude,
                        feed.getLatitude(), feed.getLongitude());
                feed.setDistance(distance);
                return distance <= radiusKm;
            })
            .sorted(Comparator.comparingDouble(Feed::getDistance))
            .limit(topN)
            .collect(Collectors.toList());
}

4.2 推荐召回 ​

推荐系统采用多路召回策略，从不同维度获取候选内容，再由精排模型统一排序。

多路召回架构 ​

用户请求推荐
    │
    ▼
┌────────────────────────────────────────────────────┐
│              多路召回并行执行                       │
│                                                    │
│  ┌─────────┐ ┌──────────┐ ┌───────┐ ┌──────────┐  │
│  │ 热度召回 │ │ 协同召回  │ │标签召回│ │ LBS 召回 │  │
│  │ Top 500 │ │ Top 200  │ │Top 200│ │ Top 200  │  │
│  └────┬────┘ └────┬─────┘ └───┬───┘ └─────┬────┘  │
│       │           │            │            │       │
│  ┌────┴───────────┴────────────┴────────────┴┐      │
│  │           社交关系召回                      │      │
│  │         (好友互动过的内容)                  │      │
│  │              Top 200                      │      │
│  └────────────────┬──────────────────────────┘      │
└───────────────────┼────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────┐
│                 结果融合去重                         │
│           权重配比 + 去重 + 保底策略                  │
│            候选集大小：500 ~ 1000                    │
└──────────────────────┬──────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────┐
│                 精排模型排序                         │
│            LR / GBDT / Wide&Deep                    │
│             CTR 预估 + 得分排序                      │
└──────────────────────┬──────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────┐
│             重排（多样性控制）                       │
│          MMR / 类目打散 / 去重                       │
│            最终结果：Top 20                         │
└─────────────────────────────────────────────────────┘
java
// 多路召回服务
@Service
public class MultiChannelRecallService {

    @Autowired
    private HotRecallChannel hotChannel;
    @Autowired
    private CollaborativeRecallChannel cfChannel;
    @Autowired
    private TagRecallChannel tagChannel;
    @Autowired
    private LBSRecallChannel lbsChannel;
    @Autowired
    private SocialRecallChannel socialChannel;

    // 线程池并行执行多路召回
    private final ExecutorService recallExecutor = Executors.newFixedThreadPool(6);

    public List<RecallItem> recall(RecallContext context) {
        List<CompletableFuture<List<RecallItem>>> futures = new ArrayList<>();

        // 1. 并行执行各路召回
        futures.add(CompletableFuture.supplyAsync(
                () -> hotChannel.recall(context), recallExecutor));
        futures.add(CompletableFuture.supplyAsync(
                () -> cfChannel.recall(context), recallExecutor));
        futures.add(CompletableFuture.supplyAsync(
                () -> tagChannel.recall(context), recallExecutor));
        futures.add(CompletableFuture.supplyAsync(
                () -> lbsChannel.recall(context), recallExecutor));
        futures.add(CompletableFuture.supplyAsync(
                () -> socialChannel.recall(context), recallExecutor));

        // 2. 合并去重
        Map<Long, RecallItem> merged = new LinkedHashMap<>();
        for (CompletableFuture<List<RecallItem>> future : futures) {
            List<RecallItem> items = future.join();
            for (RecallItem item : items) {
                merged.merge(item.getItemId(), item, (oldItem, newItem) -> {
                    // 同一条内容出现在多路召回中，累加召回得分
                    oldItem.setRecallScore(oldItem.getRecallScore() + newItem.getRecallScore());
                    oldItem.addRecallSources(newItem.getRecallSources());
                    return oldItem;
                });
            }
        }

        // 3. 保底策略：如果召回结果不足，用热门内容补充
        if (merged.size() < MIN_CANDIDATE_SIZE) {
            List<RecallItem> backupItems = hotChannel.recallWithExclude(
                    context, MIN_CANDIDATE_SIZE - merged.size(), merged.keySet());
            for (RecallItem item : backupItems) {
                merged.putIfAbsent(item.getItemId(), item);
            }
        }

        return new ArrayList<>(merged.values());
    }
}

各路召回实现 ​

java
// 热度召回
@Component
public class HotRecallChannel implements RecallChannel {
    @Override
    public List<RecallItem> recall(RecallContext context) {
        // 从热门排行 ZSET 中取 Top 500
        Set<ZSetOperations.TypedTuple<String>> topItems =
                redisTemplate.opsForZSet()
                        .reverseRangeWithScores(RedisKey.HOT_FEED_RANK, 0, 499);

        return topItems.stream()
                .map(tuple -> {
                    RecallItem item = new RecallItem(Long.parseLong(tuple.getValue()));
                    item.setRecallScore(tuple.getScore());
                    item.setRecallChannel("hot");
                    return item;
                })
                .collect(Collectors.toList());
    }
}

// 协同召回
@Component
public class CollaborativeRecallChannel implements RecallChannel {
    @Override
    public List<RecallItem> recall(RecallContext context) {
        List<Long> cfItems = userCFRecommender.recommendByUsers(
                context.getUserId(), 100);
        List<Long> itemCfItems = new ArrayList<>();
        // 找用户最近互动的物品做 ItemCF 召回
        List<Long> recentItems = interactMapper.getRecentInteractedItems(
                context.getUserId(), 5);
        for (Long itemId : recentItems) {
            itemCfItems.addAll(
                    itemCFRecommender.recommendByItems(context.getUserId(), itemId, 40));
        }

        // 合并 UserCF 和 ItemCF 结果
        List<RecallItem> result = new ArrayList<>();
        cfItems.forEach(id -> {
            RecallItem item = new RecallItem(id);
            item.setRecallChannel("user_cf");
            result.add(item);
        });
        itemCfItems.forEach(id -> {
            RecallItem item = new RecallItem(id);
            item.setRecallChannel("item_cf");
            result.add(item);
        });
        return result;
    }
}

4.3 排序模型 ​

召回后的候选内容需要经过排序模型进行 CTR 预估，按预估值排序后展示给用户。

特征工程 ​

排序模型使用的特征分为三类：

特征类别特征示例说明
用户特征用户 ID、年龄、性别、活跃度、兴趣标签向量、历史 CTR、近 7 天互动数描述用户画像和近期行为
物品特征内容 ID、作者 ID、内容类型、发布时间、标签向量、历史点击率、完播率描述内容属性
上下文特征当前时间（时段）、网络类型、设备类型、地理位置、页面位置描述请求上下文
java
// 特征提取
public FeatureVector extractFeatures(Long userId, Long itemId, RecommendContext ctx) {
    FeatureVector fv = new FeatureVector();

    // 用户特征
    UserProfile user = userProfileMapper.selectById(userId);
    fv.addFeature("user_age", user.getAge());
    fv.addFeature("user_gender", user.getGender());
    fv.addFeature("user_activeness", user.getActivenessScore()); // 30 天活跃度
    fv.addFeature("user_7d_interact_count", interactMapper.count7d(userId));
    fv.addFeature("user_following_count", followMapper.countFollows(userId));

    // 物品特征
    Feed feed = feedMapper.selectById(itemId);
    fv.addFeature("item_type", feed.getContentType());
    fv.addFeature("item_age_hours", (System.currentTimeMillis() - feed.getCreateTime().getTime()) / 3600000.0);
    fv.addFeature("item_author_followers", followMapper.countFollowers(feed.getUserId()));
    fv.addFeature("item_like_rate", calcLikeRate(feed));     // 点赞率
    fv.addFeature("item_comment_rate", calcCommentRate(feed)); // 评论率

    // 上下文特征
    fv.addFeature("ctx_hour", LocalTime.now().getHour());
    fv.addFeature("ctx_is_weekend", LocalDate.now().getDayOfWeek().getValue() >= 6 ? 1 : 0);
    fv.addFeature("ctx_platform", ctx.getPlatform()); // ios / android / web
    fv.addFeature("ctx_network", ctx.getNetworkType()); // wifi / 4g / 5g

    // 交叉特征
    // 用户与作者的互动强度
    fv.addFeature("cross_user_author_interact", interactMapper.countInteract(userId, feed.getUserId(), 30));
    // 用户对该类型内容的偏好
    fv.addFeature("cross_user_type_prefer", interactMapper.countByType(userId, feed.getContentType(), 30));

    return fv;
}

LR（逻辑回归） ​

作为排序模型基线，LR 模型可解释性强、训练速度快。

java
// LR 模型预测 CTR
public class LRRankingModel {

    // 模型权重（从离线训练结果加载）
    private Map<String, Double> weights = loadWeights();
    private double bias = 0.0;

    public double predictCTR(FeatureVector fv) {
        double score = bias;
        for (Map.Entry<String, Double> feature : fv.getFeatures().entrySet()) {
            Double weight = weights.getOrDefault(feature.getKey(), 0.0);
            score += weight * feature.getValue();
        }
        // Sigmoid 函数转换为 0~1 概率
        return 1.0 / (1.0 + Math.exp(-score));
    }
}

GBDT（梯度提升树） ​

GBDT 能自动学习特征之间的非线性关系和交叉组合。

python
# GBDT 模型训练（Python 伪代码）
import lightgbm as lgb

# 训练数据
train_data = lgb.Dataset(
    data=X_train,  # 特征矩阵
    label=y_train, # CTR
    categorical_feature=['user_gender', 'item_type', 'ctx_platform', 'ctx_network']
)

# 参数配置
params = {
    'boosting_type': 'gbdt',
    'objective': 'binary',
    'metric': 'auc',
    'num_leaves': 63,
    'learning_rate': 0.05,
    'feature_fraction': 0.8,
    'bagging_fraction': 0.8,
    'bagging_freq': 5,
    'verbose': 0,
    'num_threads': 8
}

# 训练
model = lgb.train(
    params,
    train_data,
    num_boost_round=500,
    valid_sets=[train_data],
    callbacks=[lgb.early_stopping(50)]
)

# 特征重要性分析
importance = model.feature_importance()
feature_names = model.feature_name()
for name, imp in sorted(zip(feature_names, importance), key=lambda x: x[1], reverse=True)[:10]:
    print(f"{name}: {imp}")

Wide & Deep ​

Wide & Deep 模型结合了线性模型的记忆能力和深度神经网络的泛化能力。

Wide 部分：lr(w^T * x + b)             Deep 部分：DNN(embedding + hidden layers)
       │                                        │
       └────────────────┬───────────────────────┘
                        ▼
                ┌────────────────┐
                │    Sigmoid     │
                │   CTR 预估     │
                └────────────────┘
python
# Wide & Deep 模型定义（TensorFlow）
import tensorflow as tf

def build_wide_deep_model(feature_columns):
    # Wide 部分：稀疏特征 + 交叉特征
    wide = tf.keras.layers.DenseFeatures(
        feature_columns['wide'])(inputs['wide_input'])

    # Deep 部分：稠密特征经 Embedding + MLP
    deep = tf.keras.layers.DenseFeatures(
        feature_columns['deep'])(inputs['deep_input'])
    deep = tf.keras.layers.Dense(256, activation='relu')(deep)
    deep = tf.keras.layers.BatchNormalization()(deep)
    deep = tf.keras.layers.Dropout(0.3)(deep)
    deep = tf.keras.layers.Dense(128, activation='relu')(deep)
    deep = tf.keras.layers.Dropout(0.3)(deep)
    deep = tf.keras.layers.Dense(64, activation='relu')(deep)

    # 拼接 Wide + Deep
    combined = tf.keras.layers.concatenate([wide, deep])
    output = tf.keras.layers.Dense(1, activation='sigmoid')(combined)

    model = tf.keras.Model(inputs=inputs, outputs=output)
    model.compile(
        optimizer='adam',
        loss='binary_crossentropy',
        metrics=['auc', 'accuracy']
    )
    return model

特征工程 ​

特征工程是排序模型的核心，直接影响模型效果上限。

特征处理流程：
原始数据 → 预处理(缺失值/异常值) → 标准化(归一化/离散化)
    → 特征交叉(笛卡尔积/Hash) → 特征选择(VIF/卡方检验)
    → 训练样本(TFRecord/Parquet)
处理步骤方法说明
缺失值处理均值填充 / 众数填充 / -1 占位连续特征用均值，类别特征用众数
异常值处理分位数截断 / 3-sigma超过 3 倍标准差的值截断到边界
标准化Z-score / Min-MaxLR 和 DNN 需要标准化，GBDT 不需要
特征交叉二阶交叉 / FM用户性别 x 内容类别 等组合
特征选择方差过滤 / 互信息 / L1 正则去除低方差的冗余特征
java
// 特征预处理
public class FeaturePreprocessor {

    // 连续特征标准化（Z-score）
    public double normalizeContinuous(double value, double mean, double std) {
        if (std == 0) return 0;
        return (value - mean) / std;
    }

    // 类别特征编码（频次编码）
    public double encodeCategorical(String categoryValue,
                                     Map<String, Integer> freqMap, int total) {
        // 低频类别统一映射为 0
        int freq = freqMap.getOrDefault(categoryValue, 0);
        if (freq < MIN_CATEGORY_FREQ) return 0;
        return (double) freq / total;
    }

    // 特征交叉
    public String crossFeatures(String f1, String f2) {
        // 使用 Hash 交叉避免维度爆炸
        String raw = f1 + "_" + f2;
        return "cross_" + Math.abs(raw.hashCode()) % CROSS_HASH_BUCKETS;
    }
}

4.4 推荐冷启动 ​

新用户或新内容缺乏历史行为数据，需要特殊的冷启动策略。

新用户冷启动 ​

新用户冷启动分为三个阶段：

阶段策略说明
注册期注册标签选择用户注册时选择兴趣标签（至少选 3 个）
引导期热门推荐 + 标签匹配前 3 天主要推荐热门内容+标签匹配内容
探索期逐步增加协同召回收集到足够互动信号后过渡到完整推荐
java
// 新用户冷启动推荐
@Service
public class ColdStartService {

    public List<Feed> coldStartRecommend(Long userId, RecommendContext ctx) {
        UserProfile user = userProfileMapper.selectById(userId);
        int daysSinceRegister = calcDaysSinceRegister(user.getCreateTime());

        if (daysSinceRegister <= 3) {
            // 引导期：热门推荐 + 标签匹配
            return coldStartStage(userId, ctx);
        } else {
            // 正常推荐（冷启动结束）
            return normalRecommend(userId, ctx);
        }
    }

    private List<Feed> coldStartStage(Long userId, RecommendContext ctx) {
        List<Feed> result = new ArrayList<>();

        // 1. 根据注册标签推荐（50%）
        List<UserTag> registerTags = tagMapper.selectUserTags(userId);
        if (!registerTags.isEmpty()) {
            List<Feed> tagFeeds = feedMapper.selectByTags(
                    registerTags.stream().map(UserTag::getTagId).collect(Collectors.toList()),
                    20);
            result.addAll(tagFeeds);
        }

        // 2. 热门内容补充（40%）
        List<Feed> hotFeeds = feedMapper.selectHotFeeds(16);
        result.addAll(hotFeeds);

        // 3. 新用户引导内容（10%）：新手指南、热门话题
        List<Feed> guideFeeds = feedMapper.selectGuideFeeds(4);
        result.addAll(guideFeeds);

        // 4. 去重并打散
        return diversify(result, 20);
    }

    // 多样性打散（MMR 算法）
    private List<Feed> diversify(List<Feed> candidates, int topN) {
        List<Feed> selected = new ArrayList<>();
        Set<Long> selectedIds = new HashSet<>();
        Map<Integer, Integer> categoryCount = new HashMap<>();

        for (Feed feed : candidates) {
            if (selected.size() >= topN) break;
            if (selectedIds.contains(feed.getFeedId())) continue;

            // 同一类别最多出现 4 条
            int catCount = categoryCount.getOrDefault(feed.getContentType(), 0);
            if (catCount >= 4) continue;

            selected.add(feed);
            selectedIds.add(feed.getFeedId());
            categoryCount.merge(feed.getContentType(), 1, Integer::sum);
        }

        return selected;
    }
}

新内容冷启动 ​

新发布的内容缺乏互动数据，需要给予初始曝光机会。

java
// 新内容保底曝光
@Component
public class NewItemBoostStrategy {

    // 新内容在发布后的 N 小时内获得初始曝光
    private static final int BOOST_HOURS = 2;
    private static final int BOOST_WEIGHT = 100; // 初始热度权重

    @Scheduled(fixedRate = 60000) // 每分钟执行
    public void boostNewItems() {
        List<Feed> newFeeds = feedMapper.selectFeedsSince(
                System.currentTimeMillis() - BOOST_HOURS * 3600000L);

        for (Feed feed : newFeeds) {
            // 检查是否已有足够互动（互动数 > 10 则不再需要保底）
            int interactCount = interactMapper.countTotalInteract(feed.getFeedId());
            if (interactCount < 10) {
                // 给予初始热度权重
                redisTemplate.opsForZSet().add(
                        RedisKey.HOT_FEED_RANK,
                        String.valueOf(feed.getFeedId()),
                        BOOST_WEIGHT);
            }
        }
    }
}

注册标签引导 ​

用户在注册时选择兴趣标签，系统据此提供初始推荐。

java
// 注册标签引导
public List<Tag> getRegisterTags() {
    // 返回一级标签（保证覆盖面广，易于理解）
    return tagMapper.selectLevel1Tags();
}

public void saveUserRegisterTags(Long userId, List<Integer> tagIds) {
    // 保存用户标签，初始权重 1.0
    for (Integer tagId : tagIds) {
        UserTag userTag = new UserTag();
        userTag.setUserId(userId);
        userTag.setTagId(tagId);
        userTag.setWeight(1.0);
        userTag.setSource("register");
        tagMapper.insertUserTag(userTag);
    }
    // 立即刷新推荐缓存
    cacheManager.evict(RedisKey.userRecommendCache(userId));
}

5. AB 实验 ​

5.1 实验分层 ​

AB 实验系统采用多层正交架构，支持同时运行多个实验。

流量分层模型 ​

总流量 (100%)
    │
    ├──── Layer 1: 推荐算法层
    │      ├── Experiment A: 新协同过滤算法 (10%)
    │      ├── Experiment B: 新排序模型 (10%)
    │      ├── Experiment C: 新召回策略 (10%)
    │      └── Control: 基线 (70%)
    │
    ├──── Layer 2: UI 交互层
    │      ├── Experiment D: 新 Feed 样式 (10%)
    │      ├── Experiment E: 新弹幕样式 (10%)
    │      └── Control: 基线 (80%)
    │
    ├──── Layer 3: 礼物策略层
    │      ├── Experiment F: 新礼物价格 (10%)
    │      └── Control: 基线 (90%)
    │
    └──── Layer 4: 匹配策略层
           ├── Experiment G: 新匹配权重 (10%)
           └── Control: 基线 (90%)

各层之间流量正交（通过 Hash 算法实现），同一用户可以同时参与多个层的实验。

java
// 流量分层与分组
@Component
public class ExperimentService {

    // 分层 Hash 分配
    public ExperimentGroup assignGroup(Long userId, String layerName) {
        // 使用用户 ID + 层名做 Hash，保证层间正交
        String hashInput = userId + "_" + layerName;
        int hash = Math.abs(hashInput.hashCode()) % 100;

        // 根据层配置获取实验分组
        LayerConfig layer = layerConfigMapper.selectByName(layerName);
        for (Experiment exp : layer.getExperiments()) {
            if (hash >= exp.getStartPercent() && hash < exp.getEndPercent()) {
                return new ExperimentGroup(exp.getExpId(), exp.getExpName());
            }
        }
        return new ExperimentGroup(0, "control"); // 默认对照组
    }

    // 获取用户在所有层的实验分组
    public Map<String, ExperimentGroup> getAllGroups(Long userId) {
        List<String> layerNames = layerConfigMapper.selectAllLayerNames();
        Map<String, ExperimentGroup> result = new HashMap<>();
        for (String layerName : layerNames) {
            result.put(layerName, assignGroup(userId, layerName));
        }
        return result;
    }
}

参数配置 ​

实验参数通过配置中心动态下发，支持实时调整。

yaml
# 实验参数配置（Apollo / Nacos）
experiments:
  feed_recall:
    control:
      hot_weight: 0.3
      cf_weight: 0.3
      tag_weight: 0.2
      lbs_weight: 0.1
      social_weight: 0.1
    exp_new_recall:
      hot_weight: 0.2
      cf_weight: 0.4
      tag_weight: 0.2
      lbs_weight: 0.1
      social_weight: 0.1

  ranking_model:
    control:
      model: lr
      lr_learning_rate: 0.01
    exp_deep_model:
      model: wide_deep
      deep_learning_rate: 0.001
      deep_hidden_units: [256, 128, 64]

  live_danmu:
    control:
      max_danmu_per_sec: 3
      danmu_display_count: 10
    exp_more_danmu:
      max_danmu_per_sec: 5
      danmu_display_count: 15

指标采集 ​

实验指标通过埋点系统实时采集，分为事件埋点和业务埋点。

java
// 实验埋点采集
@Component
public class ExperimentTracker {

    // 埋点事件发送
    public void trackEvent(ExperimentEvent event) {
        // 补充实验分组信息
        Map<String, ExperimentGroup> groups = experimentService.getAllGroups(event.getUserId());
        event.setGroups(groups);

        // 异步发送到消息队列（批量聚合后入 HBase / ClickHouse）
        mqTemplate.convertAndSend(MQTopic.EXPERIMENT_TRACK, event);
    }

    // 曝光埋点
    public void trackImpression(Long userId, Long feedId, String page) {
        ExperimentEvent event = new ExperimentEvent();
        event.setEventType("impression");
        event.setUserId(userId);
        event.setItemId(feedId);
        event.setPage(page);
        event.setTimestamp(System.currentTimeMillis());
        trackEvent(event);
    }

    // 点击埋点
    public void trackClick(Long userId, Long feedId, String page) {
        ExperimentEvent event = new ExperimentEvent();
        event.setEventType("click");
        event.setUserId(userId);
        event.setItemId(feedId);
        event.setPage(page);
        event.setTimestamp(System.currentTimeMillis());
        trackEvent(event);
    }

    // 时长埋点
    public void trackDuration(Long userId, Long feedId, long durationMs) {
        ExperimentEvent event = new ExperimentEvent();
        event.setEventType("duration");
        event.setUserId(userId);
        event.setItemId(feedId);
        event.setValue(durationMs);
        event.setTimestamp(System.currentTimeMillis());
        trackEvent(event);
    }
}

5.2 实验评估指标 ​

AB 实验的评估指标体系覆盖用户参与度、内容消费质量和商业价值。

核心指标定义 ​

指标计算方式说明
CTR（点击率）点击次数 / 曝光次数衡量推荐准确性
使用时长单次会话平均时长衡量用户粘性
次日留存率次日回访用户数 / 当日活跃用户数衡量产品价值
互动率(点赞+评论+转发+收藏) / 曝光次数衡量内容质量
人均 Feed 消费数总 Feed 曝光数 / 活跃用户数衡量内容消费深度
送礼转化率送礼用户数 / 直播观看用户数衡量付费转化
java
// 实验指标计算
@Component
public class ExperimentMetricsCalculator {

    // 计算 CTR
    public double calcCTR(Long expId, Date startTime, Date endTime) {
        long impressions = trackMapper.countEvents(expId, "impression", startTime, endTime);
        long clicks = trackMapper.countEvents(expId, "click", startTime, endTime);
        return impressions == 0 ? 0 : (double) clicks / impressions;
    }

    // 计算平均使用时长（分钟）
    public double calcAvgDuration(Long expId, Date startTime, Date endTime) {
        List<SessionDuration> sessions = trackMapper.getSessions(expId, startTime, endTime);
        return sessions.stream()
                .mapToLong(SessionDuration::getDurationMs)
                .average()
                .orElse(0) / 60000.0;
    }

    // 计算次日留存率
    public double calcDayRetention(Long expId, Date day) {
        Date nextDay = DateUtils.addDays(day, 1);
        // 当日活跃用户
        long activeUsers = trackMapper.countDistinctUsers(expId, day);
        // 次日回访用户
        long retainedUsers = trackMapper.countDistinctUsers(expId, nextDay);
        return activeUsers == 0 ? 0 : (double) retainedUsers / activeUsers;
    }

    // 计算互动率
    public double calcInteractRate(Long expId, Date startTime, Date endTime) {
        long impressions = trackMapper.countEvents(expId, "impression", startTime, endTime);
        // 互动行为汇总（这里简化处理，实际需区分不同互动类型）
        long interactions = interactMapper.countInteractionsByExp(expId, startTime, endTime);
        return impressions == 0 ? 0 : (double) interactions / impressions;
    }
}

统计显著性检验 ​

实验评估需要统计显著性检验，确保观测到的差异不是随机波动。

java
// 统计显著性检验（t 检验）
public class StatisticalValidator {

    // 判断实验结果是否显著
    public ExperimentResult evaluate(ExperimentMetrics control, ExperimentMetrics experiment) {
        double ctrlMean = control.getMean();
        double expMean = experiment.getMean();
        double ctrlVar = control.getVariance();
        double expVar = experiment.getVariance();
        long ctrlN = control.getSampleSize();
        long expN = experiment.getSampleSize();

        // 双样本 t 检验
        double se = Math.sqrt(ctrlVar / ctrlN + expVar / expN);
        if (se == 0) return new ExperimentResult(false, 0, "样本量不足");

        double tStat = (expMean - ctrlMean) / se;
        // 近似计算 p 值（自由度取较小样本量）
        double df = Math.min(ctrlN, expN) - 1;
        double pValue = 2 * (1 - studentTCdf(Math.abs(tStat), df));

        // 判断置信度
        boolean significant = pValue < 0.05; // 95% 置信度

        String summary = String.format(
                "CTRL: %.4f | EXP: %.4f | Lift: %.2f%% | p-value: %.4f | %s",
                ctrlMean, expMean, (expMean - ctrlMean) / ctrlMean * 100,
                pValue, significant ? "显著" : "不显著");

        return new ExperimentResult(significant, pValue, summary);
    }

    // 学生 t 分布 CDF（近似计算）
    private double studentTCdf(double t, double df) {
        // 使用 Beta 函数近似
        double x = df / (df + t * t);
        return 1 - 0.5 * betaInc(x, df / 2, 0.5);
    }

    private double betaInc(double x, double a, double b) {
        // 使用正则化不完全 Beta 函数的近似实现
        // 实际生产环境可使用 Apache Commons Math 库
        return new org.apache.commons.math3.special.Beta()
                .regularizedBeta(x, a, b);
    }
}

实验报告 ​

java
// 实验报告生成
@Service
public class ExperimentReportService {

    public Report generateReport(Long expId, Date startTime, Date endTime) {
        Report report = new Report();
        report.setExpId(expId);
        report.setStartTime(startTime);
        report.setEndTime(endTime);

        // 获取实验组和对照组的指标
        Experiment exp = experimentMapper.selectById(expId);
        Long controlExpId = exp.getControlExpId();

        // 核心指标对比
        report.addMetric("CTR", calcMetric(controlExpId, expId, this::calcCTR));
        report.addMetric("平均使用时长(分钟)", calcMetric(controlExpId, expId, this::calcAvgDuration));
        report.addMetric("次日留存率", calcMetric(controlExpId, expId, this::calcDayRetention));
        report.addMetric("互动率", calcMetric(controlExpId, expId, this::calcInteractRate));

        // 结论
        report.setConclusion(drawConclusion(report));
        return report;
    }

    private String drawConclusion(Report report) {
        long positiveCount = report.getMetrics().stream()
                .filter(m -> m.isSignificant() && m.getLift() > 0)
                .count();
        long negativeCount = report.getMetrics().stream()
                .filter(m -> m.isSignificant() && m.getLift() < 0)
                .count();

        if (positiveCount > negativeCount) {
            return "实验组效果优于对照组，建议全量发布";
        } else if (negativeCount > positiveCount) {
            return "实验组效果劣于对照组，建议下线实验";
        } else {
            return "实验组与对照组无显著差异，建议延长实验周期";
        }
    }
}

实验决策流程 ​

提出假设 → 设计实验 → 分配流量 → 采集数据
    → 定期评估（每日/每周）
        ├── 指标显著正向 → 全量发布
        ├── 指标显著负向 → 下线实验
        └── 指标不显著 → 继续实验 / 调整参数 / 增加样本量

AB 实验平台应当支持以下能力：

  • 流量分层与正交：同一用户可参与多个实验，实验间互不干扰
  • 实时指标看板：实验启动后实时展示关键指标变化曲线
  • 自动决策告警：指标出现显著负向时自动告警并暂停实验
  • 累计效应分析：评估实验的长期影响（如 7 日/14 日留存）
  • 网络效应隔离：社交类实验需要考虑实验组对对照组的影响溢出