视频评论数、收藏数通过mq、redis实现

xiangpei

6 天以前 1cdb060a8aa59b0979f7609db1781805528e76e7

 framework/src/main/java/cn/lili/modules/lmk/service/impl/VideoCommentServiceImpl.java

@@ -6,11 +6,17 @@
import cn.lili.common.security.AuthUser;
import cn.lili.common.security.context.UserContext;
import cn.lili.common.sensitive.SensitiveWordsFilter;
import cn.lili.modules.lmk.constant.RedisKeyExpireConstant;
import cn.lili.modules.lmk.domain.entity.ThumbsUpRecord;
import cn.lili.modules.lmk.domain.form.ThumbsUpRecordForm;
import cn.lili.modules.lmk.domain.vo.CollectTypeNumVO;
import cn.lili.modules.lmk.enums.general.VideoCommentStatusEnum;
import cn.lili.modules.lmk.event.event.VideoCommentNumCacheEvent;
import cn.lili.modules.lmk.service.ThumbsUpRecordService;
import cn.lili.rocketmq.RocketmqSendCallbackBuilder;
import cn.lili.rocketmq.tags.CommentTagsEnum;
import cn.lili.rocketmq.tags.OrderTagsEnum;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.metadata.IPage;
import cn.lili.modules.lmk.domain.entity.VideoComment;
import cn.lili.modules.lmk.mapper.VideoCommentMapper;
@@ -22,8 +28,12 @@
import cn.lili.modules.lmk.domain.form.VideoCommentForm;
import cn.lili.modules.lmk.domain.vo.VideoCommentVO;
import cn.lili.modules.lmk.domain.query.VideoCommentQuery;
import org.apache.commons.collections4.ListUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import lombok.RequiredArgsConstructor;
import cn.lili.utils.PageUtil;
@@ -51,11 +61,7 @@
    private final ThumbsUpRecordService thumbsUpRecordService;
    private final RocketMQTemplate rocketMQTemplate;
    private final RocketmqCustomProperties rocketmqCustomProperties;

    /**
     * 评论点赞数的过期时间
     */
    public final static long EXPIRE_TIME = 15l;
    private final ApplicationEventPublisher eventPublisher;

    /**
     * 添加
@@ -84,7 +90,9 @@
        entity.setUserAvatar(currentUser.getFace());
        baseMapper.insert(entity);
        // 初始化redis中评论的点赞数量
        cache.put(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(entity.getId()), 0, EXPIRE_TIME, TimeUnit.DAYS);
        cache.put(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(entity.getId()), 0, RedisKeyExpireConstant.COMMENT_LIKE_NUM_EXPIRE, RedisKeyExpireConstant.EXPIRE_DAY);
        // 处理视频评论数
        eventPublisher.publishEvent(new VideoCommentNumCacheEvent(this, entity.getVideoId()));
        return Result.ok("添加成功").data(this.detail(entity.getId()).get("data"));
    }

@@ -176,56 +184,107 @@

    @Override
    public Result thumbsUp(ThumbsUpRecordForm form) {
        boolean exists = new LambdaQueryChainWrapper<>(thumbsUpRecordService.getBaseMapper())
                .eq(ThumbsUpRecord::getRefId, form.getRefId())
                .eq(ThumbsUpRecord::getUserId, UserContext.getCurrentUserId())
                .exists();
        if (exists) {
            return Result.ok();
        }
        ThumbsUpRecord record = ThumbsUpRecordForm.getEntityByForm(form, null);
        record.setUserId(UserContext.getCurrentUserId());
        thumbsUpRecordService.save(record);
        if (cache.exist(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(record.getRefId()))) {
            cache.incr(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(record.getRefId()));
        } else {
            VideoComment comment = baseMapper.selectById(form.getRefId());
            if (Objects.nonNull(comment)) {
                cache.put(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(comment.getId()), comment.getThumbsUpNum() + 1, EXPIRE_TIME, TimeUnit.DAYS);
            }
        }
        // TODO 使用rocketmq异步写入到mysql中
//        rocketMQTemplate.asyncSend();

        // 走mq异步处理
        String destination = rocketmqCustomProperties.getCommentTopic() + ":" + CommentTagsEnum.THUMBS_UP.name();
        rocketMQTemplate.asyncSend(destination, JSON.toJSONString(record), RocketmqSendCallbackBuilder.commonCallback());
        return Result.ok();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void mqThumbsUp(ThumbsUpRecord record) {
        boolean exists = new LambdaQueryChainWrapper<>(thumbsUpRecordService.getBaseMapper())
                .eq(ThumbsUpRecord::getRefId, record.getRefId())
                .eq(ThumbsUpRecord::getUserId, record.getUserId())
                .exists();
        if (exists) {
            return;
        }
        thumbsUpRecordService.save(record);
        VideoComment comment = this.getById(record.getRefId());
        if (cache.exist(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(record.getRefId()))) {
            cache.incr(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(record.getRefId()));
        } else {
            if (Objects.nonNull(comment)) {
                cache.put(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(comment.getId()), comment.getThumbsUpNum() + 1, RedisKeyExpireConstant.COMMENT_LIKE_NUM_EXPIRE, RedisKeyExpireConstant.EXPIRE_DAY);
            }
        }
        // 标识该评论需要通过定时任务统计点赞数
        if (Objects.nonNull(comment) && ! comment.getThumbsUpJob()) {
            new LambdaUpdateChainWrapper<>(baseMapper)
                    .eq(VideoComment::getId, comment.getId())
                    .set(VideoComment::getThumbsUpJob, Boolean.TRUE)
                    .update();
        }
    }

    @Override
    public Result cancelThumbsUp(ThumbsUpRecordForm form) {
        new LambdaUpdateChainWrapper<>(thumbsUpRecordService.getBaseMapper())
                .eq(ThumbsUpRecord::getRefId, form.getRefId())
                .eq(ThumbsUpRecord::getThumbsUpType, form.getThumbsUpType())
                .eq(ThumbsUpRecord::getUserId, UserContext.getCurrentUserId())
                .remove();
        // redis数量减一
        cache.decr(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(form.getRefId()));
        // TODO mq异步同步到mysql
        ThumbsUpRecord record = ThumbsUpRecordForm.getEntityByForm(form, null);
        record.setUserId(UserContext.getCurrentUserId());
        // 走mq异步处理
        String destination = rocketmqCustomProperties.getCommentTopic() + ":" + CommentTagsEnum.CANCEL_THUMBS_UP.name();
        rocketMQTemplate.asyncSend(destination, JSON.toJSONString(record), RocketmqSendCallbackBuilder.commonCallback());
        return Result.ok();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void mqCancelThumbsUp(ThumbsUpRecord record) {
        new LambdaUpdateChainWrapper<>(thumbsUpRecordService.getBaseMapper())
                .eq(ThumbsUpRecord::getRefId, record.getRefId())
                .eq(ThumbsUpRecord::getThumbsUpType, record.getThumbsUpType())
                .eq(ThumbsUpRecord::getUserId, record.getUserId())
                .remove();
        // redis数量减一
        VideoComment comment = this.getById(record.getRefId());
        if (cache.exist(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(record.getRefId()))) {
            cache.decr(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(record.getRefId()));
        } else {
            if (Objects.nonNull(comment)) {
                cache.put(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(comment.getId()), comment.getThumbsUpNum() - 1, RedisKeyExpireConstant.COMMENT_LIKE_NUM_EXPIRE, RedisKeyExpireConstant.EXPIRE_DAY);
            }
        }
        // 标识该评论需要通过定时任务统计点赞数
        if (Objects.nonNull(comment) && ! comment.getThumbsUpJob()) {
            new LambdaUpdateChainWrapper<>(baseMapper)
                    .eq(VideoComment::getId, comment.getId())
                    .set(VideoComment::getThumbsUpJob, Boolean.TRUE)
                    .update();
        }
    }

    /**
     * 从redis中获取评论数量，如果redis中没有则将mysql中的数量写入到redis
     * 从redis中获取评论点赞数量，如果redis中没有则将mysql中的数量写入到redis
     *
     * @param commentId
     * @param mysqlNum
     * @return
     */
    private long getCommentThumbsUpNum(String commentId, long mysqlNum) {
    private Integer getCommentThumbsUpNum(String commentId, Integer mysqlNum) {
        Object redisNum = cache.get(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(commentId));
        if (Objects.isNull(redisNum)) {
            // redis中没有就把数据库的写到redis中
            cache.put(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(commentId), mysqlNum, EXPIRE_TIME, TimeUnit.DAYS);
            cache.put(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(commentId), mysqlNum, RedisKeyExpireConstant.COMMENT_LIKE_NUM_EXPIRE, RedisKeyExpireConstant.EXPIRE_DAY);
            return mysqlNum;
        }
        return Long.valueOf((Integer) redisNum);
        return (Integer) redisNum;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateCommentThumbsUpNumBatch(List<CollectTypeNumVO> numList) {
        // 按500条数据进行拆分
        List<List<CollectTypeNumVO>> chunks = ListUtils.partition(numList, 500);
        for (List<CollectTypeNumVO> chunk : chunks) {
            baseMapper.updateCommentThumbsUpNumBatch(chunk);
            new LambdaUpdateChainWrapper<>(baseMapper)
                    .in(VideoComment::getId, chunk.stream().map(CollectTypeNumVO::getId).collect(Collectors.toList()))
                    .set(VideoComment::getThumbsUpJob, Boolean.FALSE)
                    .update();
        }
    }
}