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

1. Feed 流系统

1.1 Feed 流分类

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

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

时间线 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

基于用户的地理位置（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 (超低延迟)                      │
└────────────────────────────────────────┘

推流协议对比

协议	延迟	优点	缺点
RTMP	2~5s	成熟稳定，生态好	基于 TCP，弱网表现一般
SRT	0.5~2s	弱网抗丢包强，支持 AES 加密	较新，部分 CDN 不支持
WebRTC	<500ms	超低延迟，P2P 能力	大规模分发成本高
HLS	5~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);
    }
}

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

1. Feed 流系统 ​

1.1 Feed 流分类 ​

时间线 Feed ​

社交推荐 Feed ​

热门排行 Feed ​

同城 Feed ​

1.2 Feed 推送模式对比 ​

推模式（写扩散） ​

拉模式（读扩散） ​

推拉混合模式 ​

1.3 Feed 存储设计 ​

Feed 内容表 ​

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

拉取游标分页 ​

1.4 Feed 热度排序 ​

热度算法公式 ​

定期重算 ​

2. Feed 流架构 ​

2.1 发布扇出架构 ​

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

2.3 Feed 快慢路径 ​

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

慢路径（大 V 拉模式） ​

混合路径的收件箱补充 ​

3. 直播互动 ​

3.1 直播间架构 ​

整体架构 ​

推流协议对比 ​

推流接入示例 ​

CDN 转码配置 ​

3.2 直播间实时互动 ​

弹幕系统（WebSocket） ​

礼物打赏 ​

连麦 PK ​

3.3 礼物系统 ​

礼物表设计 ​

充值流程 ​

3.4 直播间状态管理 ​

状态定义 ​

状态转换 ​

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

1. Feed 流系统

1.1 Feed 流分类

时间线 Feed

社交推荐 Feed

热门排行 Feed

同城 Feed

1.2 Feed 推送模式对比

推模式（写扩散）

拉模式（读扩散）

推拉混合模式

1.3 Feed 存储设计

Feed 内容表

时间线表（粉丝收件箱）

拉取游标分页

1.4 Feed 热度排序

热度算法公式

定期重算

2. Feed 流架构

2.1 发布扇出架构

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

2.3 Feed 快慢路径

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

慢路径（大 V 拉模式）

混合路径的收件箱补充

3. 直播互动

3.1 直播间架构

整体架构

推流协议对比

推流接入示例

CDN 转码配置

3.2 直播间实时互动

弹幕系统（WebSocket）

礼物打赏

连麦 PK

3.3 礼物系统

礼物表设计

充值流程

3.4 直播间状态管理

状态定义

状态转换