实时推送

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

@@ -39,7 +39,7 @@ public class LiveCompletionPointsTask {
     private ILiveService liveService;
 
     /**
-     * 定时检查观看时长并创建完课记录
+     * 定时检查观看时长并创建完课记录（兜底机制）
      * 每分钟执行一次
      * 优化：防重复推送 + 只查询开启了完课积分的直播间
      */
@@ -81,9 +81,6 @@ public class LiveCompletionPointsTask {
                             
                             completionPointsRecordService.checkAndCreateCompletionRecord(liveId, userId, duration);
 
-                            // 5. 检查是否有新的完课记录待领取,推送弹窗消息（防重复）
-                            sendCompletionNotificationOnce(liveId, userId);
-
                         } catch (Exception e) {
                             log.error("处理用户完课状态失败, liveId={}, userId={}", liveId, entry.getKey(), e);
                         }
@@ -98,30 +95,4 @@ public class LiveCompletionPointsTask {
             log.error("检查完课状态定时任务执行失败", e);
         }
     }
-
-   private void sendCompletionNotificationOnce(Long liveId, Long userId) {
-    try {
-        // 查询未领取的完课记录
-        List<LiveCompletionPointsRecord> unreceivedRecords = 
-            completionPointsRecordService.getUserUnreceivedRecords(liveId, userId);
-        
-        if (unreceivedRecords == null || unreceivedRecords.isEmpty()) {
-            return;
-        }
-
-        SendMsgVo sendMsgVo = new SendMsgVo();
-        sendMsgVo.setLiveId(liveId);
-        sendMsgVo.setUserId(userId);
-        sendMsgVo.setCmd("completionPoints");
-        sendMsgVo.setMsg("完成任务!");
-        sendMsgVo.setData(JSONObject.toJSONString(unreceivedRecords.get(0)));
-        
-        webSocketServer.sendCompletionPointsMessage(liveId, userId, sendMsgVo);
-        
-        log.info("发送完课积分弹窗通知, liveId={}, userId={}, points={}", 
-                liveId, userId, unreceivedRecords.get(0).getPointsAwarded());
-        } catch (Exception e) {
-        log.error("发送完课通知失败", e);
-        }
-    }
 }

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

@@ -36,6 +36,7 @@ import javax.websocket.*;
 import javax.websocket.server.ServerEndpoint;
 import java.io.EOFException;
 import java.io.IOException;
+import java.time.LocalDateTime;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.locks.Lock;
@@ -80,6 +81,7 @@ public class WebSocketServer {
     private final LiveCouponMapper liveCouponMapper = SpringUtils.getBean(LiveCouponMapper.class);
     private final ILiveWatchLogService liveWatchLogService = SpringUtils.getBean(ILiveWatchLogService.class);
     private final ILiveVideoService liveVideoService = SpringUtils.getBean(ILiveVideoService.class);
+    private final ILiveCompletionPointsRecordService completionPointsRecordService = SpringUtils.getBean(ILiveCompletionPointsRecordService.class);
     private static Random random = new Random();
     
     // Redis key 前缀：用户进入直播间时间
@@ -349,6 +351,32 @@ public class WebSocketServer {
                     if (msg.getData() != null && !msg.getData().isEmpty()) {
                         try {
                             Long currentDuration = Long.parseLong(msg.getData());
+
+                            Live currentLive = liveService.selectLiveByLiveId(liveId);
+                            if (currentLive == null) {
+                                break;
+                            }
+                            
+                            // 判断直播是否已开始：status=2(直播中) 或 当前时间 >= 开播时间
+                            boolean isLiveStarted = false;
+                            if (currentLive.getStatus() != null && currentLive.getStatus() == 2) {
+                                // status=2 表示直播中
+                                isLiveStarted = true;
+                            } else if (currentLive.getStartTime() != null) {
+                                // 判断当前时间是否已超过开播时间
+                                LocalDateTime now = java.time.LocalDateTime.now();
+                                isLiveStarted = now.isAfter(currentLive.getStartTime()) || now.isEqual(currentLive.getStartTime());
+                            }
+                            
+                            if (!isLiveStarted) {
+                                log.debug("[心跳-观看时长] 直播未开始（开播倒计时中），不统计观看时长, liveId={}, status={}, startTime={}", 
+                                        liveId, currentLive.getStatus(), currentLive.getStartTime());
+                                break;
+                            }
+                            
+                            log.debug("[心跳-观看时长] 直播已开始，统计观看时长, liveId={}, userId={}, duration={}秒", 
+                                    liveId, watchUserId, currentDuration);
+                            
                             // 使用Hash结构存储：一个直播间一个Hash，包含所有用户的时长
                             String hashKey = "live:watch:duration:hash:" + liveId;
                             String userIdField = String.valueOf(watchUserId);
@@ -361,6 +389,8 @@ public class WebSocketServer {
                                 // 设置过期时间（2小时）
                                 redisCache.expire(hashKey, 2, TimeUnit.HOURS);
 
+                                checkAndSendCompletionPointsInRealTime(liveId, watchUserId, currentDuration);
+
                             }
                         } catch (Exception e) {
                             log.error("[心跳-观看时长] 更新失败, liveId={}, userId={}, data={}", 
@@ -1315,5 +1345,48 @@ public class WebSocketServer {
         }
     }
 
+    /**
+     * 实时检查并推送完课积分
+     * 在用户观看时长更新时立即检查是否达到完课条件，达到则立即推送
+     * @param liveId 直播间ID
+     * @param userId 用户ID
+     * @param duration 当前观看时长（秒）
+     */
+    private void checkAndSendCompletionPointsInRealTime(long liveId, long userId, Long duration) {
+        try {
+            log.debug("[实时完课检查] liveId={}, userId={}, duration={}秒", liveId, userId, duration);
+
+            // 1. 调用完课记录服务检查并创建完课记录
+            completionPointsRecordService.checkAndCreateCompletionRecord(liveId, userId, duration);
+
+            // 2. 查询是否有新的未领取完课记录
+            List<LiveCompletionPointsRecord> unreceivedRecords =
+                completionPointsRecordService.getUserUnreceivedRecords(liveId, userId);
+
+            if (unreceivedRecords == null || unreceivedRecords.isEmpty()) {
+                // 没有待领取的完课记录
+                return;
+            }
+
+            // 3. 构建推送消息
+            SendMsgVo sendMsgVo = new SendMsgVo();
+            sendMsgVo.setLiveId(liveId);
+            sendMsgVo.setUserId(userId);
+            sendMsgVo.setCmd("completionPoints");
+            sendMsgVo.setMsg("完成任务!");
+            sendMsgVo.setData(JSONObject.toJSONString(unreceivedRecords.get(0)));
+
+            // 4. 实时推送完课积分弹窗
+            sendCompletionPointsMessage(liveId, userId, sendMsgVo);
+
+            log.info("[实时完课推送] 发送完课积分弹窗通知, liveId={}, userId={}, points={}, duration={}秒",
+                    liveId, userId, unreceivedRecords.get(0).getPointsAwarded(), duration);
+
+        } catch (Exception e) {
+            log.error("[实时完课推送] 实时检查完课积分失败, liveId={}, userId={}, duration={}",
+                    liveId, userId, duration, e);
+        }
+    }
+
 }
 

fs-service/src/main/java/com/fs/his/mapper/FsIntegralCartMapper.java

@@ -28,4 +28,11 @@ public interface FsIntegralCartMapper extends BaseMapper<FsIntegralCart> {
      * @return  list
      */
     List<FsIntegralCartVO> getCartsByMap(@Param("params") Map<String, Object> params);
+
+    /**
+     * 根据商品ID删除购物车记录
+     * @param goodsId   商品ID
+     * @return  删除数量
+     */
+    int deleteCartByGoodsId(@Param("goodsId") Long goodsId);
 }

fs-service/src/main/java/com/fs/his/service/impl/FsIntegralGoodsServiceImpl.java

@@ -9,6 +9,7 @@ import com.fs.course.domain.FsCourseAnswerReward;
 import com.fs.his.domain.FsChineseMedicine;
 import com.fs.his.domain.FsIntegralGoods;
 import com.fs.his.domain.FsUser;
+import com.fs.his.mapper.FsIntegralCartMapper;
 import com.fs.his.mapper.FsIntegralGoodsMapper;
 import com.fs.his.mapper.FsUserMapper;
 import com.fs.his.param.FsIntegralGoodsListUParam;
@@ -37,6 +38,8 @@ public class FsIntegralGoodsServiceImpl implements IFsIntegralGoodsService
     private FsIntegralGoodsMapper fsIntegralGoodsMapper;
     @Autowired
     private FsUserMapper fsUserMapper;
+    @Autowired
+    private FsIntegralCartMapper fsIntegralCartMapper;
 
     @Autowired
     private ISysConfigService configService;
@@ -99,6 +102,11 @@ public class FsIntegralGoodsServiceImpl implements IFsIntegralGoodsService
     @Override
     public int deleteFsIntegralGoodsByGoodsIds(Long[] goodsIds)
     {
+        // 先删除购物车中的相关记录
+        for (Long goodsId : goodsIds) {
+            fsIntegralCartMapper.deleteCartByGoodsId(goodsId);
+        }
+        // 再删除商品
         return fsIntegralGoodsMapper.deleteFsIntegralGoodsByGoodsIds(goodsIds);
     }
 
@@ -111,6 +119,9 @@ public class FsIntegralGoodsServiceImpl implements IFsIntegralGoodsService
     @Override
     public int deleteFsIntegralGoodsByGoodsId(Long goodsId)
     {
+        // 先删除购物车中的相关记录
+        fsIntegralCartMapper.deleteCartByGoodsId(goodsId);
+        // 再删除商品
         return fsIntegralGoodsMapper.deleteFsIntegralGoodsByGoodsId(goodsId);
     }
 

fs-service/src/main/resources/mapper/his/FsIntegralCartMapper.xml

@@ -41,6 +41,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         select
             ifnull(sum(ic.cart_num), 0)
         from fs_integral_cart ic
+        inner join fs_integral_goods ig on ig.goods_id = ic.goods_id
         <where>
             <if test="params.userId != null">
                 ic.user_id = #{params.userId}
@@ -79,4 +80,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
             </if>
         </where>
     </select>
+
+    <delete id="deleteCartByGoodsId">
+        delete from fs_integral_cart where goods_id = #{goodsId}
+    </delete>
 </mapper>