第 246 章:综合项目实战 - 短视频平台

学习目标

• 视频上传与转码
• 推荐系统核心
• 点赞评论关注
• Feed 流设计

一、系统架构

二、视频上传

2.1 分片上传

@RestController
@RequestMapping("/api/video/upload")
public class VideoUploadController {

    @Autowired
    private VideoUploadService uploadService;

    @Autowired
    private OssClient ossClient;

    @PostMapping("/init")
    public InitUploadResponse init(@RequestBody InitUploadRequest req) {
        String uploadId = uploadService.initUpload(
            req.getUserId(),
            req.getFileName(),
            req.getFileSize(),
            req.getChunkSize()
        );
        return new InitUploadResponse(uploadId);
    }

    @PostMapping("/part")
    public UploadPartResponse uploadPart(@RequestParam String uploadId,
                                          @RequestParam Integer partNumber,
                                          @RequestParam MultipartFile chunk) throws IOException {
        String etag = uploadService.uploadPart(uploadId, partNumber, chunk);
        return new UploadPartResponse(partNumber, etag);
    }

    @PostMapping("/complete")
    public CompleteUploadResponse complete(@RequestBody CompleteUploadRequest req) {
        String url = uploadService.completeUpload(
            req.getUploadId(),
            req.getParts()
        );
        return new CompleteUploadResponse(url);
    }
}

@Service
@Slf4j
public class VideoUploadService {

    public String initUpload(Long userId, String fileName, long fileSize, int chunkSize) {
        // 计算分片数
        int partCount = (int) Math.ceil(fileSize * 1.0 / chunkSize);

        // OSS 初始化
        String objectKey = "videos/" + userId + "/" + UUID.randomUUID();
        InitiateMultipartUploadRequest initReq = new InitiateMultipartUploadRequest(BUCKET, objectKey);
        String uploadId = ossClient.initiateMultipartUpload(initReq).getUploadId();

        // 保存到 DB
        VideoUpload upload = new VideoUpload();
        upload.setUploadId(uploadId);
        upload.setUserId(userId);
        upload.setFileName(fileName);
        upload.setObjectKey(objectKey);
        upload.setFileSize(fileSize);
        upload.setPartCount(partCount);
        upload.setStatus(UploadStatus.UPLOADING);
        uploadMapper.insert(upload);

        return uploadId;
    }

    public String uploadPart(String uploadId, int partNumber, MultipartFile chunk) throws IOException {
        VideoUpload upload = uploadMapper.findByUploadId(uploadId);

        UploadPartRequest partReq = new UploadPartRequest();
        partReq.setBucketName(BUCKET);
        partReq.setKey(upload.getObjectKey());
        partReq.setUploadId(uploadId);
        partReq.setPartNumber(partNumber);
        partReq.setInputStream(chunk.getInputStream());
        partReq.setPartSize(chunk.getSize());

        UploadPartResult result = ossClient.uploadPart(partReq);

        // 保存分片信息
        uploadPartMapper.savePart(uploadId, partNumber, result.getETag());

        return result.getETag();
    }

    public String completeUpload(String uploadId, List<PartETag> parts) {
        VideoUpload upload = uploadMapper.findByUploadId(uploadId);

        CompleteMultipartUploadRequest completeReq = new CompleteMultipartUploadRequest(
            BUCKET, upload.getObjectKey(), uploadId, parts
        );
        ossClient.completeMultipartUpload(completeReq);

        // 更新状态
        upload.setStatus(UploadStatus.UPLOADED);
        uploadMapper.updateById(upload);

        // 异步触发转码
        kafkaTemplate.send("video-uploaded", upload.getId());

        return "https://cdn.example.com/" + upload.getObjectKey();
    }
}

2.2 视频转码

@Component
@Slf4j
public class VideoTranscoder {

    @KafkaListener(topics = "video-uploaded", groupId = "transcoder")
    public void onUploaded(Long videoId) {
        Video video = videoService.findById(videoId);

        try {
            // 提交转码任务
            String jobId = submitTranscodeJob(video);

            // 等待转码完成
            TranscodeResult result = waitForCompletion(jobId);

            // 保存转码结果
            video.setTranscoded(true);
            video.setHlsUrl(result.getHlsUrl());
            video.setThumbnailUrl(result.getThumbnailUrl());
            video.setDuration(result.getDuration());
            videoService.update(video);

            // 触发 AI 审核
            kafkaTemplate.send("video-audit", videoId);

        } catch (Exception e) {
            video.setStatus(VideoStatus.TRANSCODE_FAILED);
            videoService.update(video);
        }
    }
}

三、Feed 流设计

3.1 推模式 vs 拉模式

我们采用 推拉结合 方案。

3.2 Feed 服务

@Service
public class FeedService {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private UserFollowingService followService;

    private static final int FEED_MAX_SIZE = 1000;

    /**
     * 推模式:活跃用户视频写入粉丝 Feed
     */
    public void pushToFollowers(Long userId, Video video) {
        List<Long> followers = followService.getFollowers(userId, 0, 10000);

        for (Long follower : followers) {
            String feedKey = "feed:" + follower;
            // 用 LPUSH + LTRIM 保持最近 1000 条
            redisTemplate.opsForList().leftPush(feedKey, JSON.toJSONString(video));
            redisTemplate.opsForList().trim(feedKey, 0, FEED_MAX_SIZE - 1);
        }
    }

    /**
     * 拉模式兜底:不活跃用户主动拉取
     */
    public List<Video> pullFeed(Long userId, int page, int size) {
        // 1. 先查 Redis
        String feedKey = "feed:" + userId;
        List<Video> cached = redisTemplate.opsForList()
            .range(feedKey, (long)(page-1)*size, (long)page*size - 1)
            .stream()
            .map(s -> JSON.parseObject(s, Video.class))
            .collect(Collectors.toList());

        if (!cached.isEmpty()) {
            return cached;
        }

        // 2. 不活跃:从 ES 聚合
        return elasticsearchService.queryByFollowings(userId, page, size);
    }

    /**
     * 混合方案 - 取活跃用户推送的 + 不活跃关注的
     */
    public List<Video> getFeed(Long userId, int page, int size) {
        // 活跃等级
        int activity = userActivityService.getLevel(userId);

        if (activity > 80) {
            return pullFeed(userId, page, size);
        } else if (activity > 30) {
            return pullFeed(userId, page, size);
        } else {
            // 不活跃,主动拉
            return elasticsearchService.queryByFollowings(userId, page, size);
        }
    }
}

四、互动功能

4.1 点赞

@Service
@Slf4j
public class LikeService {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private KafkaTemplate<String, LikeEvent> kafkaTemplate;

    /**
     * 点赞(Lua 保证幂等)
     */
    public boolean like(Long userId, Long videoId) {
        String key = "like:" + videoId;
        String userKey = "user:like:" + userId;

        String luaScript = """
            if redis.call('SISMEMBER', KEYS[2], ARGV[1]) == 1 then
                return -1
            end
            redis.call('SADD', KEYS[2], ARGV[1])
            redis.call('INCR', KEYS[1])
            return 1
            """;

        DefaultRedisScript<Long> script = new DefaultRedisScript<>(luaScript, Long.class);
        Long result = redisTemplate.execute(script,
            List.of(key, userKey),
            userId.toString()
        );

        if (result != null && result == 1) {
            kafkaTemplate.send("video-liked", new LikeEvent(userId, videoId));
            return true;
        }
        return false;
    }

    /**
     * 是否点赞
     */
    public boolean isLiked(Long userId, Long videoId) {
        return Boolean.TRUE.equals(
            redisTemplate.opsForSet().isMember("user:like:" + userId, videoId.toString())
        );
    }

    /**
     * 点赞数
     */
    public long countLikes(Long videoId) {
        String count = redisTemplate.opsForValue().get("like:count:" + videoId);
        return count != null ? Long.parseLong(count) : 0;
    }
}

4.2 评论

@Service
public class CommentService {

    public Long addComment(Long userId, Long videoId, String content, Long parentId) {
        Comment comment = new Comment();
        comment.setUserId(userId);
        comment.setVideoId(videoId);
        comment.setContent(content);
        comment.setParentId(parentId);
        comment.setCreatedAt(LocalDateTime.now());
        commentMapper.insert(comment);

        // 更新评论数
        redisTemplate.opsForValue().increment("comment:count:" + videoId);

        // 通知被评论者
        if (parentId != null) {
            Comment parent = commentMapper.selectById(parentId);
            notificationService.notify(parent.getUserId(),
                "你的评论收到新回复",
                comment.getId()
            );
        }

        // 推送到视频作者
        Video video = videoService.findById(videoId);
        if (!video.getUserId().equals(userId)) {
            notificationService.notify(video.getUserId(),
                "你的视频收到新评论",
                comment.getId()
            );
        }

        return comment.getId();
    }
}

五、推荐系统

5.1 双塔模型 + 向量召回

5.2 召回 + 排序

@Service
public class RecommendService {

    public List<Video> recommend(Long userId, int size) {
        // 1. 多路召回
        List<RecallResult> candidates = new ArrayList<>();

        // 1.1 双塔召回
        candidates.addAll(doubleTowerRecall(userId, 500));

        // 1.2 协同过滤
        candidates.addAll(itemCfRecall(userId, 200));

        // 1.3 热门
        candidates.addAll(hotRecall(100));

        // 1.4 关注的人
        candidates.addAll(followRecall(userId, 100));

        // 2. 去重合并
        Map<Long, RecallResult> merged = mergeAndDedup(candidates);

        // 3. 精排(调用模型服务)
        List<RankRequest> rankRequests = merged.values().stream()
            .map(r -> buildRankRequest(userId, r))
            .collect(Collectors.toList());

        List<RankResult> ranked = modelService.rank(rankRequests);

        // 4. 重排(规则调整)
        return rerank(userId, ranked, size);
    }
}

六、视频审核

@Component
@Slf4j
public class ContentAuditService {

    @KafkaListener(topics = "video-audit", groupId = "audit")
    public void audit(Long videoId) {
        Video video = videoService.findById(videoId);

        try {
            // 1. 文本审核(标题、描述)
            TextScanResponse textResp = greenClient.textScan(video.getTitle(), video.getDescription());

            // 2. 图片审核(封面)
            ImageScanResponse imgResp = greenClient.imageScan(video.getThumbnailUrl());

            // 3. 视频审核(异步)
            String jobId = greenClient.videoScanAsync(video.getHlsUrl());

            video.setAuditJobId(jobId);
            video.setStatus(VideoStatus.AUDITING);
            videoService.update(video);

        } catch (Exception e) {
            log.error("审核提交失败", e);
        }
    }

    @KafkaListener(topics = "video-audit-result", groupId = "audit-result")
    public void handleResult(AuditResultMessage message) {
        Video video = videoService.findById(message.getVideoId());

        if (message.isPass()) {
            video.setStatus(VideoStatus.PUBLISHED);
        } else {
            video.setStatus(VideoStatus.REJECTED);
            video.setRejectReason(message.getReason());
            // 通知创作者
            notificationService.notifyReject(video.getUserId(), message.getReason());
        }

        videoService.update(video);
    }
}

七、播放器优化

7.1 自适应码率

hls:
  resolutions:
    - 240p: 400kbps
    - 480p: 1000kbps
    - 720p: 2500kbps
    - 1080p: 5000kbps

CDN:
  边缘节点:
    - 全球 200+ 节点
    - 智能调度
    - 命中率 > 95%

7.2 预加载

// Web 端
const preload = (nextVideos) => {
    nextVideos.forEach(video => {
        const link = document.createElement('link');
        link.rel = 'preload';
        link.as = 'video';
        link.href = video.url;
        document.head.appendChild(link);
    });
};

// 滑动到 80% 时预加载下一个
videoContainer.addEventListener('scroll', (e) => {
    const scrollPercent = e.target.scrollTop / e.target.scrollHeight;
    if (scrollPercent > 0.8) {
        preload(upcomingVideos);
    }
});

八、本章小结

模块	关键
上传	分片 + OSS + 断点续传
存储	对象存储 + CDN + 转码
Feed	推拉结合 + Redis + ES
推荐	多路召回 + 精排 + 重排
互动	Lua 幂等 + MQ 异步

动手练习

1. 实现一个简化版的分片上传 demo
2. 设计 Feed 流的推拉结合方案
3. 用 Redis Set 实现点赞系统
4. 模拟推荐召回的数据流

---

下一章:下一章:第 247 章:综合项目实战 - 即时通讯