KafkaStream流式实时计算，对聚合后的消息进行消费处理

adlx

1068人浏览 · 2025-08-03 18:41:23

adlx · 2025-08-03 18:41:23 发布

@Component
@Slf4j
public class ArticleIncrHandleListener {

    @Autowired
    private ApArticleService apArticleService;

    @KafkaListener(topics = HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC)
    public void onMessage(String mess){
        if(StringUtils.isNotBlank(mess)){
            ArticleVisitStreamMess articleVisitStreamMess = JSON.parseObject(mess, ArticleVisitStreamMess.class);
            apArticleService.updateScore(articleVisitStreamMess);

        }
    }
}
/**
 * 更新文章的分值  同时更新缓存中的热点文章数据
 * @param mess
 */
@Override
public void updateScore(ArticleVisitStreamMess mess) {
    //1.更新文章的阅读、点赞、收藏、评论的数量
    ApArticle apArticle = updateArticle(mess);
    //2.计算文章的分值
    Integer score = computeScore(apArticle);
    score = score * 3;

    //3.替换当前文章对应频道的热点数据
    replaceDataToRedis(apArticle, score, ArticleConstants.HOT_ARTICLE_FIRST_PAGE + apArticle.getChannelId());

    //4.替换推荐对应的热点数据
    replaceDataToRedis(apArticle, score, ArticleConstants.HOT_ARTICLE_FIRST_PAGE + ArticleConstants.DEFAULT_TAG);

}

/**
 * 替换数据并且存入到redis
 * @param apArticle
 * @param score
 * @param s
 */
private void replaceDataToRedis(ApArticle apArticle, Integer score, String s) {
    String articleListStr = cacheService.get(s);
    if (StringUtils.isNotBlank(articleListStr)) {
        List<HotArticleVo> hotArticleVoList = JSON.parseArray(articleListStr, HotArticleVo.class);

        boolean flag = true;

        //如果缓存中存在该文章，只更新分值
        for (HotArticleVo hotArticleVo : hotArticleVoList) {
            if (hotArticleVo.getId().equals(apArticle.getId())) {
                hotArticleVo.setScore(score);
                flag = false;
                break;
            }
        }

        //如果缓存中不存在，查询缓存中分值最小的一条数据，进行分值的比较，如果当前文章的分值大于缓存中的数据，就替换
        if (flag) {
            if (hotArticleVoList.size() >= 30) {
                hotArticleVoList = hotArticleVoList.stream().sorted(Comparator.comparing(HotArticleVo::getScore).reversed()).collect(Collectors.toList());
                HotArticleVo lastHot = hotArticleVoList.get(hotArticleVoList.size() - 1);
                if (lastHot.getScore() < score) {
                    hotArticleVoList.remove(lastHot);
                    HotArticleVo hot = new HotArticleVo();
                    BeanUtils.copyProperties(apArticle, hot);
                    hot.setScore(score);
                    hotArticleVoList.add(hot);
                }


            } else {
                HotArticleVo hot = new HotArticleVo();
                BeanUtils.copyProperties(apArticle, hot);
                hot.setScore(score);
                hotArticleVoList.add(hot);
            }
        }
        //缓存到redis
        hotArticleVoList = hotArticleVoList.stream().sorted(Comparator.comparing(HotArticleVo::getScore).reversed()).collect(Collectors.toList());
        cacheService.set(s, JSON.toJSONString(hotArticleVoList));

    }
}

/**
 * 更新文章行为数量
 * @param mess
 */
private ApArticle updateArticle(ArticleVisitStreamMess mess) {
    ApArticle apArticle = getById(mess.getArticleId());
    apArticle.setCollection(apArticle.getCollection()==null?0:apArticle.getCollection()+mess.getCollect());
    apArticle.setComment(apArticle.getComment()==null?0:apArticle.getComment()+mess.getComment());
    apArticle.setLikes(apArticle.getLikes()==null?0:apArticle.getLikes()+mess.getLike());
    apArticle.setViews(apArticle.getViews()==null?0:apArticle.getViews()+mess.getView());
    updateById(apArticle);
    return apArticle;

}

/**
 * 计算文章的具体分值
 * @param apArticle
 * @return
 */
private Integer computeScore(ApArticle apArticle) {
    Integer scere = 0;
    if(apArticle.getLikes() != null){
        scere += apArticle.getLikes() * ArticleConstants.HOT_ARTICLE_LIKE_WEIGHT;
    }
    if(apArticle.getViews() != null){
        scere += apArticle.getViews();
    }
    if(apArticle.getComment() != null){
        scere += apArticle.getComment() * ArticleConstants.HOT_ARTICLE_COMMENT_WEIGHT;
    }
    if(apArticle.getCollection() != null){
        scere += apArticle.getCollection() * ArticleConstants.HOT_ARTICLE_COLLECTION_WEIGHT;
    }

    return scere;
}

这是一套完整的 “实时热点文章计算系统” 的后端处理逻辑，涉及 Kafka 消费、文章分值更新、Redis 缓存维护等关键操作。

我会用通俗易懂的方式，从上到下、从整体到细节，帮你彻底理解它在做什么。

🌟 整体架构图（先有个大局观）

Kafka 流处理（前文代码）
       ↓
生成实时统计消息 → {"articleId":123, "like":2, "view":5, ...}
       ↓
发送到 Topic: hot.article.incr.handle.topic
       ↓
<ArticleIncrHandleListener> 消费这条消息 ✅（就是你现在看的代码）
       ↓
调用 apArticleService.updateScore(...) 更新文章分值 + 缓存
       ↓
最终：热门文章排行榜实时刷新！

✅ 第一部分：`@Component` 和 `@KafkaListener` —— 消费消息

@Component
@Slf4j
public class ArticleIncrHandleListener {

    @Autowired
    private ApArticleService apArticleService;

    @KafkaListener(topics = HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC)
    public void onMessage(String mess){
        if(StringUtils.isNotBlank(mess)){
            ArticleVisitStreamMess articleVisitStreamMess = JSON.parseObject(mess, ArticleVisitStreamMess.class);
            apArticleService.updateScore(articleVisitStreamMess);
        }
    }
}

🔍 作用：

这是一个 Kafka 消费者，专门监听名为 HOT_ARTICLE_INCR_HANDLE_TOPIC 的 Topic。

🧩 每当 Kafka 中有新消息到来时：

把 JSON 字符串反序列化成 ArticleVisitStreamMess 对象
调用 apArticleService.updateScore(...) 方法进行处理

✅ 举例消息内容：

{
  "articleId": 123,
  "like": 2,
  "view": 5,
  "collect": 1,
  "comment": 1
}

👉 这意味着：最近10秒内，文章123被点赞2次、浏览5次、收藏1次、评论1次。

✅ 第二部分：`updateScore()` —— 核心入口方法

@Override
public void updateScore(ArticleVisitStreamMess mess) {
    //1. 更新文章的阅读、点赞、收藏、评论的数量
    ApArticle apArticle = updateArticle(mess);

    //2. 计算文章的分值
    Integer score = computeScore(apArticle);
    score = score * 3;  // 权重放大

    //3. 替换当前文章对应频道的热点数据
    replaceDataToRedis(apArticle, score, ArticleConstants.HOT_ARTICLE_FIRST_PAGE + apArticle.getChannelId());

    //4. 替换推荐对应的热点数据
    replaceDataToRedis(apArticle, score, ArticleConstants.HOT_ARTICLE_FIRST_PAGE + ArticleConstants.DEFAULT_TAG);
}

这个方法是整个“热点文章计算”的主流程，分为4步：

步骤	做了什么
1️⃣ `updateArticle`	把新来的行为数据（点赞+2等）加到数据库的文章记录上
2️⃣ `computeScore`	根据点赞、评论等数量，计算一个“热度分值”
3️⃣ & 4️⃣ `replaceDataToRedis`	把这篇文章的最新热度分值，更新到 Redis 缓存中（按频道 + 推荐）

✅ 第三部分：`updateArticle()` —— 更新数据库中的文章行为数

private ApArticle updateArticle(ArticleVisitStreamMess mess) {
    ApArticle apArticle = getById(mess.getArticleId()); // 从数据库查出文章
    apArticle.setCollection(... + mess.getCollect());
    apArticle.setComment(... + mess.getComment());
    apArticle.setLikes(... + mess.getLike());
    apArticle.setViews(... + mess.getViews());
    updateById(apArticle); // 更新回数据库
    return apArticle;
}

📌 作用：

把 Kafka 流处理传来的“增量数据”（比如点赞+2），累加到数据库中文章的总字段上。

✅ 举例：

数据库原来：likes=100
Kafka 消息：like=2
更新后：likes=102

📌 这样做是为了保证：

数据库里的总数是准确的（用于展示、统计）
下次再有消息来，可以继续累加

✅ 第四部分：`computeScore()` —— 计算文章热度分值

private Integer computeScore(ApArticle apArticle) {
    Integer scere = 0;
    if(apArticle.getLikes() != null){
        scere += apArticle.getLikes() * ArticleConstants.HOT_ARTICLE_LIKE_WEIGHT;
    }
    if(apArticle.getViews() != null){
        scere += apArticle.getViews();
    }
    if(apArticle.getComment() != null){
        scere += apArticle.getComment() * ArticleConstants.HOT_ARTICLE_COMMENT_WEIGHT;
    }
    if(apArticle.getCollection() != null){
        scere += apArticle.getCollection() * ArticleConstants.HOT_ARTICLE_COLLECTION_WEIGHT;
    }
    return scere;
}

📌 作用：

根据不同的用户行为，赋予不同权重，计算出一个“综合热度分”。

🎯 为什么要有权重？

因为不同行为对“热度”的贡献不同：

行为	权重（假设）	说明
浏览（View）	1	最常见，影响力小
点赞（Like）	3	用户表达喜欢
评论（Comment）	5	用户深度参与，价值高
收藏（Collection）	5	用户认为有价值

✅ 举例：

点赞 10 次 → 10 × 3 = 30
评论 2 次 → 2 × 5 = 10
总分 = 30 + 10 = 40

然后 score = score * 3; 是为了进一步放大差异，让热门文章更突出。

✅ 第五部分：`replaceDataToRedis()` —— 更新 Redis 缓存中的热门文章列表

这是最复杂也最重要的一步：维护 Redis 中的“热门文章排行榜”。

private void replaceDataToRedis(ApArticle apArticle, Integer score, String s) {
    String articleListStr = cacheService.get(s); // 从 Redis 读取当前榜单
    if (StringUtils.isNotBlank(articleListStr)) {
        List<HotArticleVo> hotArticleVoList = JSON.parseArray(articleListStr, HotArticleVo.class);

        boolean flag = true;

        // 1. 如果文章已在榜单中，只更新分值
        for (HotArticleVo hotArticleVo : hotArticleVoList) {
            if (hotArticleVo.getId().equals(apArticle.getId())) {
                hotArticleVo.setScore(score);
                flag = false;
                break;
            }
        }

        // 2. 如果文章不在榜单中
        if (flag) {
            if (hotArticleVoList.size() >= 30) {
                // 榜单已满，需要淘汰最差的
                hotArticleVoList = hotArticleVoList.stream()
                    .sorted(Comparator.comparing(HotArticleVo::getScore).reversed())
                    .collect(Collectors.toList());
                HotArticleVo lastHot = hotArticleVoList.get(hotArticleVoList.size() - 1); // 最后一名
                if (lastHot.getScore() < score) {
                    hotArticleVoList.remove(lastHot); // 淘汰
                    HotArticleVo newHot = new HotArticleVo();
                    BeanUtils.copyProperties(apArticle, newHot);
                    newHot.setScore(score);
                    hotArticleVoList.add(newHot); // 加入新文章
                }
            } else {
                // 榜单未满，直接加入
                HotArticleVo newHot = new HotArticleVo();
                BeanUtils.copyProperties(apArticle, newHot);
                newHot.setScore(score);
                hotArticleVoList.add(newHot);
            }
        }

        // 3. 排序后重新存回 Redis
        hotArticleVoList = hotArticleVoList.stream()
            .sorted(Comparator.comparing(HotArticleVo::getScore).reversed())
            .collect(Collectors.toList());
        cacheService.set(s, JSON.toJSONString(hotArticleVoList));
    }
}

📌 作用：

维护一个 最多30条的热门文章排行榜，支持两种榜单：

HOT_ARTICLE_FIRST_PAGE + channelId → 按频道的热门榜（如科技频道）
HOT_ARTICLE_FIRST_PAGE + DEFAULT_TAG → 全站推荐榜

🔁 更新策略：

情况	处理方式
文章已在榜中	更新它的 `score`，不改变位置
文章不在榜中，榜单未满30条	直接加入
文章不在榜中，榜单已满	比较它和最后一名的分数，如果更高就“挤进去”

🧠 为什么这么做？

避免频繁全量更新榜单
只在必要时才重新排序
保证榜单实时性 + 性能平衡

✅ 总结：这套系统到底在做什么？

模块	职责
Kafka Streams 流处理	实时统计每篇文章在10秒内的点赞、浏览等行为数量
ArticleIncrHandleListener	消费统计结果，触发更新
updateArticle	更新数据库中的文章行为总数
computeScore	计算文章热度分（加权）
replaceDataToRedis	维护 Redis 中的热门文章排行榜（按频道 + 推荐）