直播卓美

fs-common/src/main/java/com/fs/common/constant/LiveKeysConstant.java

@@ -37,6 +37,8 @@ public class LiveKeysConstant {
     public static final Integer PRODUCT_DETAIL_CACHE_EXPIRE = 300; //商品详情缓存过期时间（秒）
 
     public static final String LIVE_TAG_MARK_CACHE = "live:tag:mark:%s"; //直播间打标签缓存，存储直播间ID、开始时间和视频时长
+    //记录用户观看直播间信息 直播间id、用户id、外部联系人id、qwUserId
+    public static final String LIVE_USER_WATCH_LOG_CACHE = "live:user:watch:log:%s:%s:%s:%s";
 
 
 }

fs-live-app/src/main/java/com/fs/framework/aspectj/LiveWatchUserAspect.java

@@ -8,6 +8,9 @@ import org.aspectj.lang.JoinPoint;
 import org.aspectj.lang.annotation.AfterReturning;
 import org.aspectj.lang.annotation.Aspect;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;
 
 import java.util.Arrays;
@@ -17,11 +20,13 @@ import java.util.Set;
 @Aspect
 @Component
 @Slf4j
+@Order(Ordered.LOWEST_PRECEDENCE - 1)  // 调整切面优先级
 public class LiveWatchUserAspect {
 
 
 
     @Autowired
+    @Lazy
     private ILiveWatchUserService liveWatchUserService;
 
     @AfterReturning(pointcut = "execution(* com.fs.live.service.impl.LiveWatchUserServiceImpl.insertLiveWatchUser(..)) || " +

fs-live-app/src/main/java/com/fs/live/task/Task.java

@@ -33,6 +33,7 @@ import javax.annotation.PostConstruct;
 import java.math.BigDecimal;
 import java.time.Instant;
 import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -812,6 +813,8 @@ public class Task {
     @DistributeLock(key = "scanLiveWatchUserStatus", scene = "task")
     public void scanLiveWatchUserStatus() {
         try {
+
+            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
             // 查询所有正在直播的直播间
             List<Live> activeLives = liveService.selectNoEndLiveList();
             if (activeLives == null || activeLives.isEmpty()) {
@@ -864,7 +867,7 @@ public class Task {
                             if (onlineSeconds == null || onlineSeconds <= 0) {
                                 continue;
                             }
-                            
+
                             // 获取用户的 companyId 和 companyUserId
                             LiveUserFirstEntry liveUserFirstEntry =
                                     liveUserFirstEntryService.selectEntityByLiveIdUserIdWithCache(liveId, userId);
@@ -878,7 +881,10 @@ public class Task {
                             if (qwUserId == null || qwUserId <= 0 || externalContactId == null || externalContactId <= 0) {
                                 continue;
                             }
-
+                            //更新最新用户活跃时间
+                            String liveUserWatchLogKey = String.format(LIVE_USER_WATCH_LOG_CACHE, liveId, userId,externalContactId,qwUserId);
+                            LocalDateTime now = LocalDateTime.now();
+                            redisCache.setCacheObject(liveUserWatchLogKey,formatter.format(now),5,TimeUnit.MINUTES);
                             // 使用 updateLiveWatchLogTypeByDuration 的逻辑更新观看记录状态
                             updateLiveWatchLogTypeByDuration(liveId, userId, qwUserId, externalContactId,
                                     onlineSeconds, totalVideoDuration, updateLog);
@@ -972,6 +978,68 @@ public class Task {
         }
     }
 
+    /**
+     * 每分钟扫描一次用户在线状态用于更新用户观看记录值
+     */
+    @Scheduled(cron = "0 0/1 * * * ?")
+    @DistributeLock(key = "updateLiveWatchUserStatus", scene = "task")
+    public void updateLiveWatchUserStatus() {
+        try {
+            Set<String> keys = redisCache.redisTemplate.keys("live:user:watch:log");
+            LocalDateTime now = LocalDateTime.now();
+            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+            List<LiveWatchLog> updateLog = new ArrayList<>();
+            if (keys != null && !keys.isEmpty()) {
+                for (String key : keys) {
+                    String[] split = key.split(":");
+                    String cacheTime = redisCache.getCacheObject(key);
+                    //判断缓存的值是否已经距离现在超过一分钟
+                    if (StringUtils.isNotBlank(cacheTime)) {
+                        try {
+                            LocalDateTime cachedDateTime = LocalDateTime.parse(cacheTime, formatter);
+                            // 比较时间，判断是否超过1分钟（60秒）
+                            long secondsBetween = java.time.Duration.between(cachedDateTime, now).getSeconds();
+                            if (secondsBetween > 60) {
+                                // 距离上次记录已超过1分钟，更新状态为看课中断
+                                // 查询 LiveWatchLog
+                                LiveWatchLog queryLog = new LiveWatchLog();
+                                queryLog.setLiveId(Long.valueOf(split[0]));
+                                queryLog.setQwUserId(String.valueOf(split[3]));
+                                queryLog.setExternalContactId(Long.valueOf(split[2]));
+                                List<LiveWatchLog> logs = liveWatchLogService.selectLiveWatchLogByLogIdWithCache(queryLog);
+                                if (logs != null && !logs.isEmpty()) {
+                                    for (LiveWatchLog log : logs) {
+                                        if (log.getLogType() != null && log.getLogType() == 2) {
+                                            continue;
+                                        }
+                                        log.setLogType(4);
+                                        updateLog.add(log);
+                                    }
+                                }
+                            }
+                        } catch (Exception e) {
+                            log.error("解析缓存时间失败: cacheTime={}, error={}", cacheTime, e.getMessage());
+                        }
+                    }
+                }
+                // 批量插入回放用户数据
+                if (!updateLog.isEmpty()) {
+                    int batchSize = 500;
+                    for (int i = 0; i < updateLog.size(); i += batchSize) {
+                        int end = Math.min(i + batchSize, updateLog.size());
+                        List<LiveWatchLog> batch = updateLog.subList(i, end);
+                        liveWatchLogService.batchUpdateLiveWatchLog(batch);
+                    }
+                    for (LiveWatchLog liveWatchLog : updateLog) {
+                        redisCache.setCacheObject("live:watch:log:cache:" + liveWatchLog.getLogId(), liveWatchLog, 1, TimeUnit.HOURS);
+                    }
+                }
+            }
+        } catch (Exception ex) {
+            log.error("每分钟扫描一次用户在线状态用于更新用户观看记录值: error={}", ex.getMessage(), ex);
+        }
+    }
+
     /**
      * 批量同步Redis中的观看时长到数据库
      * 每2分钟执行一次，减少数据库压力

fs-live-app/src/main/java/com/fs/live/websocket/service/WebSocketServer.java

@@ -39,6 +39,7 @@ import java.io.IOException;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.locks.Lock;
@@ -1191,7 +1192,7 @@ public class WebSocketServer {
             queryLog.setLiveId(liveId);
             queryLog.setQwUserId(String.valueOf(qwUserId));
             queryLog.setExternalContactId(externalContactId);
-
+            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
             List<LiveWatchLog> logs = liveWatchLogService.selectLiveWatchLogList(queryLog);
             if (logs != null && !logs.isEmpty()) {
                 for (LiveWatchLog log : logs) {
@@ -1199,6 +1200,9 @@ public class WebSocketServer {
                     if (log.getLogType() == null || log.getLogType() != 2) {
                         log.setLogType(1);
                         liveWatchLogService.updateLiveWatchLog(log);
+                        String liveUserWatchLogKey = String.format(LIVE_USER_WATCH_LOG_CACHE, liveId, userId,externalContactId,qwUserId);
+                        LocalDateTime now = LocalDateTime.now();
+                        redisCache.setCacheObject(liveUserWatchLogKey,formatter.format(now),5,TimeUnit.MINUTES);
                     }
                 }
             }

fs-service/src/main/java/com/fs/company/service/impl/CompanyServiceImpl.java

@@ -50,6 +50,7 @@ import org.redisson.api.RedissonClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import com.fs.company.service.ICompanyService;
@@ -120,6 +121,7 @@ public class CompanyServiceImpl implements ICompanyService
     private TransactionTemplate transactionTemplate;
 
     @Autowired
+    @Lazy
     private ILiveService liveService;
     @Autowired
     private LiveOrderMapper liveOrderMapper;

fs-service/src/main/java/com/fs/hisStore/service/impl/FsStoreProductScrmServiceImpl.java

@@ -59,6 +59,7 @@ import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import com.fs.hisStore.service.IFsStoreProductScrmService;
@@ -121,6 +122,7 @@ public class FsStoreProductScrmServiceImpl implements IFsStoreProductScrmService
     private FsStoreProductCategoryScrmMapper fsStoreProductCategoryScrmMapper;
 
     @Autowired
+    @Lazy
     private ILiveService liveService;
 
     @Autowired

fs-service/src/main/java/com/fs/live/service/impl/LiveServiceImpl.java

@@ -57,6 +57,7 @@ import org.apache.http.impl.client.HttpClients;
 import org.apache.http.util.EntityUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import com.fs.common.utils.sign.Md5Utils;
@@ -130,6 +131,7 @@ public class LiveServiceImpl implements ILiveService
     LiveTagConfigMapper liveTagConfigMapper;
 
     @Autowired
+    @Lazy
     private ILiveWatchUserService liveWatchUserService;
 
     private static String TOKEN_VALID_CODE = "40001";

fs-service/src/main/java/com/fs/live/service/impl/LiveWatchUserServiceImpl.java

@@ -41,6 +41,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;