视频评论点赞通过mq异步添加数据库，点赞数缓存redis

xiangpei

2 天以前 654f4eebf519f015506b90d71637e6aad75e5b5c

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

@@ -6,11 +6,16 @@
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.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,7 +27,10 @@
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.stereotype.Service;
import lombok.RequiredArgsConstructor;
@@ -176,40 +184,75 @@

    @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, UserContext.getCurrentUserId())
                .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_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) {
        // 走mq异步处理
        String destination = rocketmqCustomProperties.getCommentTopic() + ":" + CommentTagsEnum.CANCEL_THUMBS_UP.name();
        rocketMQTemplate.asyncSend(destination, JSON.toJSONString(form), RocketmqSendCallbackBuilder.commonCallback());
        return Result.ok();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void mqCancelThumbsUp(ThumbsUpRecordForm form) {
        new LambdaUpdateChainWrapper<>(thumbsUpRecordService.getBaseMapper())
                .eq(ThumbsUpRecord::getRefId, form.getRefId())
                .eq(ThumbsUpRecord::getThumbsUpType, form.getThumbsUpType())
                .eq(ThumbsUpRecord::getUserId, UserContext.getCurrentUserId())
                .remove();
        // redis数量减一
        VideoComment comment = this.getById(form.getRefId());
        if (cache.exist(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(form.getRefId()))) {
        cache.decr(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(form.getRefId()));
        // TODO mq异步同步到mysql
        return Result.ok();
        } else {
            if (Objects.nonNull(comment)) {
                cache.put(CachePrefix.VIDEO_COMMENT_LIKE_NUM.getPrefixWithId(comment.getId()), comment.getThumbsUpNum() - 1, RedisKeyExpireConstant.COMMENT_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();
        }
    }

    /**
@@ -228,4 +271,18 @@
        }
        return Long.valueOf((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();
        }
    }
}