重粉判断性能优化

fs-qw-mq/src/main/java/com/fs/app/mq/RocketMQConsumerCourseRepeatByProjectService.java

@@ -0,0 +1,53 @@
+package com.fs.app.mq;
+
+import com.alibaba.fastjson.JSON;
+import com.fs.common.utils.StringUtils;
+import com.fs.course.mq.CourseRepeatByProjectMqConstants;
+import com.fs.course.param.FsUserCourseVideoAddKfUParam;
+import com.fs.course.service.ICourseRepeatByProjectService;
+import com.fs.course.vo.CourseRepeatByProjectMqMessage;
+import com.fs.his.domain.FsUser;
+import com.fs.his.mapper.FsUserMapper;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+@AllArgsConstructor
+@RocketMQMessageListener(
+        topic = CourseRepeatByProjectMqConstants.TOPIC,
+        consumerGroup = CourseRepeatByProjectMqConstants.CONSUMER_GROUP)
+public class RocketMQConsumerCourseRepeatByProjectService implements RocketMQListener<String> {
+
+    private final ICourseRepeatByProjectService courseRepeatByProjectService;
+    private final FsUserMapper fsUserMapper;
+
+    @Override
+    public void onMessage(String message) {
+        if (StringUtils.isEmpty(message)) {
+            return;
+        }
+        try {
+            CourseRepeatByProjectMqMessage mqMessage = JSON.parseObject(message, CourseRepeatByProjectMqMessage.class);
+            if (mqMessage == null || mqMessage.getParam() == null) {
+                return;
+            }
+            FsUserCourseVideoAddKfUParam param = mqMessage.getParam();
+            if (param.getUserId() == null) {
+                return;
+            }
+            FsUser fsUser = fsUserMapper.selectFsUserByUserId(param.getUserId());
+            if (fsUser == null) {
+                log.warn("看课重粉 MQ 消费：用户不存在 userId={}", param.getUserId());
+                return;
+            }
+            courseRepeatByProjectService.checkAndMarkRepeatByProject(param, fsUser);
+        } catch (Exception e) {
+            log.error("看课重粉 MQ 消费失败 message={}", message, e);
+            throw e;
+        }
+    }
+}

fs-service/src/main/java/com/fs/course/mq/CourseRepeatByProjectMqConstants.java

@@ -0,0 +1,19 @@
+package com.fs.course.mq;
+
+/**
+ * 看课按项目重粉判断 MQ 常量
+ */
+public final class CourseRepeatByProjectMqConstants {
+
+    public static final String TOPIC = "course-repeat-by-project";
+
+    public static final String CONSUMER_GROUP = "course-repeat-by-project-group";
+
+    /** 发送去重 Redis 前缀，key = prefix + userId + ":" + videoId */
+    public static final String DEDUP_KEY_PREFIX = "course:repeat:by-project:dedup:";
+
+    public static final int DEDUP_SECONDS = 60;
+
+    private CourseRepeatByProjectMqConstants() {
+    }
+}

fs-service/src/main/java/com/fs/course/service/ICourseRepeatByProjectService.java

@@ -0,0 +1,15 @@
+package com.fs.course.service;
+
+import com.fs.course.param.FsUserCourseVideoAddKfUParam;
+import com.fs.his.domain.FsUser;
+
+/**
+ * 看课按项目重粉判断与打标
+ */
+public interface ICourseRepeatByProjectService {
+
+    /**
+     * 按看课项目判断并重粉打标：同一项目下由不同销售观看视为重粉，跨项目不同销售不算重粉。
+     */
+    void checkAndMarkRepeatByProject(FsUserCourseVideoAddKfUParam param, FsUser fsUser);
+}

fs-service/src/main/java/com/fs/course/service/impl/CourseRepeatByProjectMqProducer.java

@@ -0,0 +1,74 @@
+package com.fs.course.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.fs.common.core.redis.RedisCache;
+import com.fs.course.mq.CourseRepeatByProjectMqConstants;
+import com.fs.course.param.FsUserCourseVideoAddKfUParam;
+import com.fs.course.vo.CourseRepeatByProjectMqMessage;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 看课按项目重粉：投递 MQ（由 fs-qw-mq 消费）
+ */
+@Slf4j
+@Service
+public class CourseRepeatByProjectMqProducer {
+
+    @Autowired(required = false)
+    private RocketMQTemplate rocketMQTemplate;
+
+    @Autowired
+    private RedisCache redisCache;
+
+    /**
+     * 去重后异步投递重粉判断消息
+     */
+    public void submitCheck(FsUserCourseVideoAddKfUParam param) {
+        if (param == null || param.getUserId() == null) {
+            return;
+        }
+        if (rocketMQTemplate == null) {
+            log.error("RocketMQTemplate 未注入，跳过重粉 MQ 投递 userId={}", param.getUserId());
+            return;
+        }
+        String dedupKey = buildDedupKey(param);
+        if (!redisCache.setIfAbsent(dedupKey, "1", CourseRepeatByProjectMqConstants.DEDUP_SECONDS, TimeUnit.SECONDS)) {
+            log.debug("看课重粉 MQ 去重跳过 userId={}, videoId={}", param.getUserId(), param.getVideoId());
+            return;
+        }
+
+        CourseRepeatByProjectMqMessage message = new CourseRepeatByProjectMqMessage();
+        message.setParam(param);
+
+        try {
+            String payload = JSON.toJSONString(message);
+            rocketMQTemplate.asyncSend(CourseRepeatByProjectMqConstants.TOPIC, payload, new SendCallback() {
+                @Override
+                public void onSuccess(SendResult sendResult) {
+                    log.debug("看课重粉 MQ 投递成功 userId={}, msgId={}", param.getUserId(), sendResult.getMsgId());
+                }
+
+                @Override
+                public void onException(Throwable e) {
+                    log.error("看课重粉 MQ 投递失败 userId={}, videoId={}", param.getUserId(), param.getVideoId(), e);
+                    redisCache.deleteObject(dedupKey);
+                }
+            });
+        } catch (Exception e) {
+            log.error("看课重粉 MQ 投递异常 userId={}", param.getUserId(), e);
+            redisCache.deleteObject(dedupKey);
+        }
+    }
+
+    private String buildDedupKey(FsUserCourseVideoAddKfUParam param) {
+        Long videoId = param.getVideoId() != null ? param.getVideoId() : 0L;
+        return CourseRepeatByProjectMqConstants.DEDUP_KEY_PREFIX + param.getUserId() + ":" + videoId;
+    }
+}

fs-service/src/main/java/com/fs/course/service/impl/CourseRepeatByProjectServiceImpl.java

@@ -0,0 +1,239 @@
+package com.fs.course.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.fs.common.utils.StringUtils;
+import com.fs.course.domain.*;
+import com.fs.course.mapper.FsCourseWatchLogMapper;
+import com.fs.course.mapper.FsUserCompanyQwMapper;
+import com.fs.course.mapper.FsUserCourseMapper;
+import com.fs.course.mapper.FsUserCourseVideoMapper;
+import com.fs.course.param.FsUserCourseVideoAddKfUParam;
+import com.fs.course.service.ICourseRepeatByProjectService;
+import com.fs.course.service.IFsUserCompanyBindService;
+import com.fs.course.service.IFsUserCompanyUserService;
+import com.fs.his.domain.FsUser;
+import com.fs.his.mapper.FsUserMapper;
+import com.fs.qw.domain.QwExternalContact;
+import com.fs.qw.mapper.QwExternalContactMapper;
+import com.fs.qw.service.IQwExternalContactService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Service
+public class CourseRepeatByProjectServiceImpl implements ICourseRepeatByProjectService {
+
+    @Autowired
+    private FsUserCourseMapper fsUserCourseMapper;
+    @Autowired
+    private FsUserCourseVideoMapper fsUserCourseVideoMapper;
+    @Autowired
+    private FsUserCompanyQwMapper fsUserCompanyQwMapper;
+    @Autowired
+    private IFsUserCompanyUserService userCompanyUserService;
+    @Autowired
+    private IFsUserCompanyBindService fsUserCompanyBindService;
+    @Autowired
+    private FsCourseWatchLogMapper courseWatchLogMapper;
+    @Autowired
+    private QwExternalContactMapper qwExternalContactMapper;
+    @Autowired
+    private IQwExternalContactService qwExternalContactService;
+    @Autowired
+    private FsUserMapper fsUserMapper;
+
+    @Override
+    public void checkAndMarkRepeatByProject(FsUserCourseVideoAddKfUParam param, FsUser fsUser) {
+        Long projectId = resolveCourseProjectId(param);
+        if (projectId == null || projectId == 0L) {
+            log.debug("看课重粉判断跳过：课程未关联项目，userId={}", param.getUserId());
+            return;
+        }
+        boolean projectRepeat = isRepeatWatchByProject(param, projectId);
+        boolean userAlreadyRepeat = fsUser.getIsRepeat() != null && fsUser.getIsRepeat() == 1;
+        log.info("看课重粉判断（按项目）：userId={}, projectId={}, projectRepeat={}, userAlreadyRepeat={}",
+                param.getUserId(), projectId, projectRepeat, userAlreadyRepeat);
+        if (projectRepeat || userAlreadyRepeat) {
+            markRepeatFansByProject(fsUser, param, projectId);
+            return;
+        }
+        if (param.getQwExternalId() != null) {
+            recordUserProjectBind(param, param.getQwExternalId(), "isAddKf");
+        }
+    }
+
+    private Long resolveCourseProjectId(FsUserCourseVideoAddKfUParam param) {
+        if (param.getCourseId() != null) {
+            FsUserCourse course = fsUserCourseMapper.selectFsUserCourseByCourseId(param.getCourseId());
+            if (course != null && course.getProject() != null && course.getProject() != 0L) {
+                return course.getProject();
+            }
+        }
+        if (param.getVideoId() != null) {
+            FsUserCourseVideo video = fsUserCourseVideoMapper.selectFsUserCourseVideoByVideoId(param.getVideoId());
+            if (video != null && video.getCourseId() != null) {
+                FsUserCourse course = fsUserCourseMapper.selectFsUserCourseByCourseId(video.getCourseId());
+                if (course != null && course.getProject() != null) {
+                    return course.getProject();
+                }
+            }
+        }
+        return 0L;
+    }
+
+    private boolean isRepeatWatchByProject(FsUserCourseVideoAddKfUParam param, Long projectId) {
+        Long userId = param.getUserId();
+        Long currentCompanyUserId = param.getCompanyUserId();
+        Long currentQwUserId = parseQwUserId(param.getQwUserId());
+
+        FsUserCompanyQw existBind = fsUserCompanyQwMapper.selectByUserAndProject(userId, projectId);
+        if (existBind != null) {
+            if (currentCompanyUserId != null && existBind.getCompanyUserId() != null
+                    && !existBind.getCompanyUserId().equals(currentCompanyUserId)) {
+                return true;
+            }
+            if (currentQwUserId != null && existBind.getQwUserId() != null
+                    && !existBind.getQwUserId().equals(currentQwUserId)) {
+                return true;
+            }
+        }
+
+        FsUserCompanyUser userCompanyUser = userCompanyUserService.selectByUserIdAndProjectId(userId, projectId);
+        if (userCompanyUser != null && currentCompanyUserId != null
+                && userCompanyUser.getCompanyUserId() != null
+                && !userCompanyUser.getCompanyUserId().equals(currentCompanyUserId)) {
+            return true;
+        }
+
+        List<FsUserCompanyBind> binds = fsUserCompanyBindService.list(
+                new QueryWrapper<FsUserCompanyBind>().eq("fs_user_id", userId).eq("project_id", projectId));
+        if (CollectionUtils.isNotEmpty(binds)) {
+            if (currentQwUserId != null && binds.stream().map(FsUserCompanyBind::getQwUserId).filter(Objects::nonNull)
+                    .anyMatch(qwId -> !qwId.equals(currentQwUserId))) {
+                return true;
+            }
+            if (currentCompanyUserId != null && binds.stream().map(FsUserCompanyBind::getCompanyUserId).filter(Objects::nonNull)
+                    .anyMatch(id -> !id.equals(currentCompanyUserId))) {
+                return true;
+            }
+        }
+
+        QueryWrapper<FsCourseWatchLog> watchWrapper = new QueryWrapper<FsCourseWatchLog>()
+                .eq("project", projectId)
+                .isNotNull("company_user_id");
+        watchWrapper.and(w -> {
+            w.eq("user_id", userId);
+            if (param.getQwExternalId() != null) {
+                w.or().eq("qw_external_contact_id", param.getQwExternalId());
+            }
+            return w;
+        });
+        List<FsCourseWatchLog> watchLogs = courseWatchLogMapper.selectList(watchWrapper);
+        if (CollectionUtils.isNotEmpty(watchLogs) && currentCompanyUserId != null) {
+            return watchLogs.stream().map(FsCourseWatchLog::getCompanyUserId).filter(Objects::nonNull)
+                    .anyMatch(id -> !id.equals(currentCompanyUserId));
+        }
+        return false;
+    }
+
+    private Long parseQwUserId(String qwUserId) {
+        if (StringUtils.isEmpty(qwUserId)) {
+            return null;
+        }
+        try {
+            return Long.parseLong(qwUserId);
+        } catch (NumberFormatException e) {
+            return null;
+        }
+    }
+
+    private void markRepeatFansByProject(FsUser fsUser, FsUserCourseVideoAddKfUParam param, Long projectId) {
+        List<QwExternalContact> toUpdate = new ArrayList<>();
+        if (param.getQwExternalId() != null) {
+            QwExternalContact current = qwExternalContactMapper.selectById(param.getQwExternalId());
+            if (current != null) {
+                current.setIsRepeat(1);
+                toUpdate.add(current);
+            }
+        }
+        List<FsUserCompanyBind> binds = fsUserCompanyBindService.list(
+                new QueryWrapper<FsUserCompanyBind>().eq("fs_user_id", param.getUserId()).eq("project_id", projectId));
+        if (CollectionUtils.isNotEmpty(binds)) {
+            Set<Long> contactIds = binds.stream().map(FsUserCompanyBind::getQwExternalContactId)
+                    .filter(Objects::nonNull).collect(Collectors.toSet());
+            for (Long contactId : contactIds) {
+                if (toUpdate.stream().anyMatch(c -> contactId.equals(c.getId()))) {
+                    continue;
+                }
+                QwExternalContact contact = qwExternalContactMapper.selectById(contactId);
+                if (contact != null) {
+                    contact.setIsRepeat(1);
+                    toUpdate.add(contact);
+                }
+            }
+        }
+        if (toUpdate.isEmpty()) {
+            List<QwExternalContact> noRepeatList = qwExternalContactMapper.selectList(
+                    new QueryWrapper<QwExternalContact>().eq("fs_user_id", param.getUserId()).eq("is_repeat", 0));
+            noRepeatList.forEach(e -> e.setIsRepeat(1));
+            toUpdate.addAll(noRepeatList);
+        }
+        if (!toUpdate.isEmpty()) {
+            qwExternalContactService.updateBatchById(toUpdate);
+        }
+        if (fsUser.getIsRepeat() == null || fsUser.getIsRepeat() != 1) {
+            fsUser.setIsRepeat(1);
+            fsUserMapper.updateFsUser(fsUser);
+        }
+        FsUserCompanyUser userCompanyUser = userCompanyUserService.selectByUserIdAndProjectId(param.getUserId(), projectId);
+        if (userCompanyUser != null && (userCompanyUser.getIsRepeatFans() == null || userCompanyUser.getIsRepeatFans() != 1)) {
+            userCompanyUser.setIsRepeatFans(1);
+            userCompanyUserService.updateById(userCompanyUser);
+        }
+    }
+
+    private void recordUserProjectBind(FsUserCourseVideoAddKfUParam param, Long qwExternalId, String sceneName) {
+        try {
+            FsUserCourseVideo video = fsUserCourseVideoMapper.selectFsUserCourseVideoByVideoId(param.getVideoId());
+            if (video == null || video.getCourseId() == null) {
+                log.warn("【{}-记录绑定关系】视频信息不存在，视频ID: {}", sceneName, param.getVideoId());
+                return;
+            }
+
+            FsUserCourse course = fsUserCourseMapper.selectFsUserCourseByCourseId(video.getCourseId());
+            Long projectId = course != null && course.getProject() != null ? course.getProject() : 0L;
+
+            if (projectId == 0L) {
+                log.debug("【{}-记录绑定关系】课程未关联项目，跳过记录，课程ID: {}", sceneName, video.getCourseId());
+                return;
+            }
+
+            Long qwUserId = param.getQwUserId() != null ? Long.parseLong(param.getQwUserId()) : null;
+            fsUserCompanyQwMapper.insertOrUpdate(
+                    param.getUserId(),
+                    projectId,
+                    qwUserId,
+                    param.getCompanyUserId(),
+                    param.getCompanyId(),
+                    qwExternalId,
+                    video.getCourseId(),
+                    param.getVideoId()
+            );
+
+            log.info("【{}-记录绑定关系成功】用户ID: {}, 项目ID: {}, 销售ID: {}, 外部联系人id: {}",
+                    sceneName, param.getUserId(), projectId, qwUserId, qwExternalId);
+
+        } catch (Exception e) {
+            log.error("【{}-记录绑定关系失败】用户ID: {}, 视频ID: {}, 错误: {}",
+                    sceneName, param.getUserId(), param.getVideoId(), e.getMessage(), e);
+        }
+    }
+}

fs-service/src/main/java/com/fs/course/service/impl/FsUserCourseVideoServiceImpl.java

@@ -258,6 +258,9 @@ public class FsUserCourseVideoServiceImpl extends ServiceImpl<FsUserCourseVideoM
     @Autowired
     private FsUserCompanyQwMapper fsUserCompanyQwMapper;
 
+    @Autowired
+    private CourseRepeatByProjectMqProducer courseRepeatByProjectMqProducer;
+
     @Autowired
     private SysDictDataMapper sysDictDataMapper;
 
@@ -803,32 +806,9 @@ public class FsUserCourseVideoServiceImpl extends ServiceImpl<FsUserCourseVideoM
         //非独属链接提示
         String noMemberMsg = "此链接已被绑定，请联系伴学助手领取您的专属链接，专属链接请勿分享哦!";
         try {
-            new Thread(() -> {
-                log.info("自动重粉判断开始！");
-                try {
-                    QwExternalContact qwExternalContact = qwExternalContactMapper.selectById(param.getQwExternalId());
-                    if(fsUser.getIsRepeat() == 1){
-                        List<QwExternalContact> noRepeatList = qwExternalContactMapper.selectList(new QueryWrapper<QwExternalContact>().eq("fs_user_id", param.getUserId()).eq("is_repeat", 0));
-                        noRepeatList.add(qwExternalContact);
-                        noRepeatList.forEach(e -> e.setIsRepeat(1));
-                        qwExternalContactService.updateBatchById(noRepeatList);
-                        return;
-                    }
-                    Long userId = param.getUserId();
-                    List<QwExternalContact> qwExternalContactList = qwExternalContactMapper.selectQwExternalContactByFsUserId(userId);
-                    if(qwExternalContactList !=null && qwExternalContactList.size() > 1){
-                        qwExternalContactList.add(qwExternalContact);
-                        qwExternalContactList.forEach(e -> e.setIsRepeat(1));
-                        qwExternalContactService.updateBatchById(qwExternalContactList);
-                        fsUser.setIsRepeat(1);
-                        fsUserMapper.updateFsUser(fsUser);
-                    }
-                }catch (Exception e){
-                    log.error("看课重粉判断失败", e);
-                }
-            }).start();
-        }catch (Exception e){
-            log.error("看课重粉提交mq失败", e);
+            courseRepeatByProjectMqProducer.submitCheck(param);
+        } catch (Exception e) {
+            log.error("看课重粉 MQ 投递失败", e);
         }
 
         String json = configService.selectConfigByKey("course.config");
@@ -1279,53 +1259,6 @@ public class FsUserCourseVideoServiceImpl extends ServiceImpl<FsUserCourseVideoM
         }
     }
 
-    /**
-     * 记录用户-项目-销售绑定关系到 fs_user_company_qw 表
-     * 用于企微自动发课（send_type=2）的重粉判断优化
-     *
-     * @param param 看课参数
-     * @param qwExternalId 企微外部联系人id
-     * @param sceneName 场景名称（用于日志追踪）
-     */
-    private void recordUserProjectBind(FsUserCourseVideoAddKfUParam param, Long qwExternalId, String sceneName) {
-        try {
-            // 查询视频所属的课程和项目
-            FsUserCourseVideo video = fsUserCourseVideoMapper.selectFsUserCourseVideoByVideoId(param.getVideoId());
-            if (video == null || video.getCourseId() == null) {
-                logger.warn("【{}-记录绑定关系】视频信息不存在，视频ID: {}", sceneName, param.getVideoId());
-                return;
-            }
-
-            FsUserCourse course = fsUserCourseMapper.selectFsUserCourseByCourseId(video.getCourseId());
-            Long projectId = course != null && course.getProject() != null ? course.getProject() : 0L;
-
-            // 只对有项目ID的课程记录绑定关系
-            if (projectId == 0L) {
-                logger.debug("【{}-记录绑定关系】课程未关联项目，跳过记录，课程ID: {}", sceneName, video.getCourseId());
-                return;
-            }
-
-            Long qwUserId = param.getQwUserId() != null ? Long.parseLong(param.getQwUserId()) : null;
-            fsUserCompanyQwMapper.insertOrUpdate(
-                param.getUserId(),
-                projectId,
-                qwUserId,
-                param.getCompanyUserId(),
-                param.getCompanyId(),
-                qwExternalId,
-                video.getCourseId(),
-                param.getVideoId()
-            );
-
-            logger.info("【{}-记录绑定关系成功】用户ID: {}, 项目ID: {}, 销售ID: {}, 外部联系人id: {}",
-                sceneName, param.getUserId(), projectId, qwUserId, qwExternalId);
-
-        } catch (Exception e) {
-            logger.error("【{}-记录绑定关系失败】用户ID: {}, 视频ID: {}, 错误: {}",
-                sceneName, param.getUserId(), param.getVideoId(), e.getMessage(), e);
-        }
-    }
-
     private R addCustomerService(String qwUserById,String msg){
         if ("济南联志健康".equals(signProjectName)){
             return R.error(400,msg).put("qrcode", "");

fs-service/src/main/java/com/fs/course/vo/CourseRepeatByProjectMqMessage.java

@@ -0,0 +1,17 @@
+package com.fs.course.vo;
+
+import com.fs.course.param.FsUserCourseVideoAddKfUParam;
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 看课按项目重粉 MQ 消息体
+ */
+@Data
+public class CourseRepeatByProjectMqMessage implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private FsUserCourseVideoAddKfUParam param;
+}