优化在线用户缓存逻辑

pom.xml

@@ -24,11 +24,6 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-aop</artifactId>
         </dependency>
-        <dependency>
-            <groupId>com.google.guava</groupId>
-            <artifactId>guava</artifactId>
-            <version>30.1.1-jre</version>
-        </dependency>
         <dependency>
             <groupId>org.hibernate.validator</groupId>
             <artifactId>hibernate-validator</artifactId>

src/main/java/com/upchina/common/config/cache/CacheConfig.java

@@ -0,0 +1,71 @@
+package com.upchina.common.config.cache;
+
+import com.hazelcast.config.InMemoryFormat;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static com.upchina.common.config.cache.CacheKey.*;
+
+public class CacheConfig {
+
+    public static final String DEFAULT_MAP_NAME = "default";
+
+    public static class LocalMapConfig {
+        public final int maxSize;
+        public final int liveSeconds;
+        public final InMemoryFormat inMemoryFormat;
+
+        LocalMapConfig(int maxSize, int liveSeconds) {
+            this.maxSize = maxSize;
+            this.liveSeconds = liveSeconds;
+            this.inMemoryFormat = InMemoryFormat.BINARY;
+        }
+
+        LocalMapConfig(int maxSize, int liveSeconds, InMemoryFormat inMemoryFormat) {
+            this.maxSize = maxSize;
+            this.liveSeconds = liveSeconds;
+            this.inMemoryFormat = inMemoryFormat;
+        }
+    }
+
+    public static Map<String, LocalMapConfig> getConfigMap() {
+        // 设置近地缓存实时同步,不采用批量提交策略
+        System.setProperty("hazelcast.map.invalidation.batch.enabled", "false");
+        // PER_NODE：max-size指定单个集群成员中map条目的最大数量。这是max-size的默认策略。如果使用这个配置，需要注意max-size的值必须大于分区的数量（默认为271）。
+        // PER_PARTITION：max-size指定每个分区存储的map条目最大数。这个策略建议不要在小规模的集群中使用，因为小规模的集群，单个节点包含了大量的分区，在执行回收策略时，会去按照分区的划分组个检查回收条件，导致效率低下。
+        // USED_HEAP_SIZE：指在每个Hazelcast实例中，max-size指定map所占用的内存堆的（以megabytes计算，兆字节）最大值。需要注意这个策略不能工作在in-memory-format=OBJECT，因为当数据被设置为OBJECT时，无法确定所占用的内存大小。
+        // USED_HEAP_PERCENTAGE：每个Hazelcast实例中，max-size指定map占用内存堆的百分比。例如，JVM被设置有1000MB，而这个值设置为max-size=10，当map条目数占用的堆数据超过100MB时，Hazelcast开始执行数据释放工作。需要注意的是当使用这个策略时，不能将in-memory-format设置为OBJECT，理由同上。
+        // FREE_HEAP_SIZE：max-size指定了单个JVM的堆最小空闲空间，单位为megabytes。
+        // FREE_HEAP_PERCENTAGE：max-size指定单个JVM的最小空闲空间的百分比。例如JVM分配了1000MB的空间，这个值设置为10，当空闲堆只有100MB时，会引发map的数据清除放行为。
+        // 当map条数超过10000，会引发map的数据清除行为
+        Map<String, LocalMapConfig> configMap = new HashMap<>();
+        configMap.put(DISTRIBUTED_LOCK, new LocalMapConfig(1000, 0));
+
+        configMap.put(DEPT, new LocalMapConfig(10000, 3600));
+        configMap.put(TAG, new LocalMapConfig(10000, 3600));
+        configMap.put(ADVISOR_INFO, new LocalMapConfig(10000, 300));
+
+        configMap.put(RECOMMEND, new LocalMapConfig(10000, 300));
+        configMap.put(USER, new LocalMapConfig(10000, 3600));
+        configMap.put(CAPTCHA, new LocalMapConfig(10000, 300));
+        configMap.put(RBAC, new LocalMapConfig(10000, 300));
+
+        configMap.put(URL_MAP, new LocalMapConfig(10000, 300, InMemoryFormat.OBJECT));
+        configMap.put(SCREEN, new LocalMapConfig(1000, 10, InMemoryFormat.OBJECT));
+        configMap.put(VIDEO_TX_ONLINE, new LocalMapConfig(1000, 20, InMemoryFormat.OBJECT));
+        configMap.put(VIDEO_LIVE_MESSAGE, new LocalMapConfig(10000, 86400));
+        configMap.put(VIDEO_LIVE, new LocalMapConfig(10000, 300));
+        configMap.put(VIDEO_LIVE_DELAY, new LocalMapConfig(10000, 10));
+        configMap.put(VIDEO_LIVE_COLUMN, new LocalMapConfig(10000, 300));
+        configMap.put(VIDEO_ACTIVITY, new LocalMapConfig(1000, 300));
+        configMap.put(VIDEO_LIVE_LIBRARY, new LocalMapConfig(1000, 300));
+        configMap.put(VIDEO_LIVE_USER_MAP, new LocalMapConfig(10000, 300));
+        configMap.put(CUSTOMER_MAP, new LocalMapConfig(10000, 3600));
+        configMap.put(QUESTION, new LocalMapConfig(1000, 30));
+
+        configMap.put(COURSE, new LocalMapConfig(10000, 300));
+        configMap.put(GROUP, new LocalMapConfig(10000, 300));
+        return configMap;
+    }
+}

src/main/java/com/upchina/common/config/cache/CacheKey.java

@@ -358,4 +358,8 @@ public class CacheKey {
         public static final String GROUP_MESSAGE_DATE_ID_MAP = "group_message_date_id_map|";
     }
 
+    public static final String GROUP_ONLINE_USER = "group_online_user";
+
+    public static final String VIDEO_ONLINE_USER = "video_online_user";
+
 }

src/main/java/com/upchina/common/config/cache/HazelcastConfiguration.java

@@ -8,12 +8,9 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import static com.upchina.common.config.cache.CacheKey.*;
-
 @Configuration
 public class HazelcastConfiguration {
 
@@ -23,67 +20,6 @@ public class HazelcastConfiguration {
     @Value("${hazelcast.serverPort}")
     private Integer serverPort;
 
-    private static final String DEFAULT_MAP_NAME = "default";
-
-    private static final Map<String, LocalMapConfig> configMap = new HashMap<>();
-
-    public static class LocalMapConfig {
-        private final int maxSize;
-        private final int liveSeconds;
-        private final InMemoryFormat inMemoryFormat;
-
-        LocalMapConfig(int maxSize, int liveSeconds) {
-            this.maxSize = maxSize;
-            this.liveSeconds = liveSeconds;
-            this.inMemoryFormat = InMemoryFormat.BINARY;
-        }
-
-        LocalMapConfig(int maxSize, int liveSeconds, InMemoryFormat inMemoryFormat) {
-            this.maxSize = maxSize;
-            this.liveSeconds = liveSeconds;
-            this.inMemoryFormat = inMemoryFormat;
-        }
-    }
-
-    static {
-        // 设置近地缓存实时同步,不采用批量提交策略
-        System.setProperty("hazelcast.map.invalidation.batch.enabled", "false");
-
-        // PER_NODE：max-size指定单个集群成员中map条目的最大数量。这是max-size的默认策略。如果使用这个配置，需要注意max-size的值必须大于分区的数量（默认为271）。
-        // PER_PARTITION：max-size指定每个分区存储的map条目最大数。这个策略建议不要在小规模的集群中使用，因为小规模的集群，单个节点包含了大量的分区，在执行回收策略时，会去按照分区的划分组个检查回收条件，导致效率低下。
-        // USED_HEAP_SIZE：指在每个Hazelcast实例中，max-size指定map所占用的内存堆的（以megabytes计算，兆字节）最大值。需要注意这个策略不能工作在in-memory-format=OBJECT，因为当数据被设置为OBJECT时，无法确定所占用的内存大小。
-        // USED_HEAP_PERCENTAGE：每个Hazelcast实例中，max-size指定map占用内存堆的百分比。例如，JVM被设置有1000MB，而这个值设置为max-size=10，当map条目数占用的堆数据超过100MB时，Hazelcast开始执行数据释放工作。需要注意的是当使用这个策略时，不能将in-memory-format设置为OBJECT，理由同上。
-        // FREE_HEAP_SIZE：max-size指定了单个JVM的堆最小空闲空间，单位为megabytes。
-        // FREE_HEAP_PERCENTAGE：max-size指定单个JVM的最小空闲空间的百分比。例如JVM分配了1000MB的空间，这个值设置为10，当空闲堆只有100MB时，会引发map的数据清除放行为。
-        // 当map条数超过10000，会引发map的数据清除行为
-        configMap.put(DISTRIBUTED_LOCK, new LocalMapConfig(1000, 0));
-
-        configMap.put(DEPT, new LocalMapConfig(10000, 3600));
-        configMap.put(TAG, new LocalMapConfig(10000, 3600));
-        configMap.put(ADVISOR_INFO, new LocalMapConfig(10000, 300));
-
-        configMap.put(RECOMMEND, new LocalMapConfig(10000, 300));
-        configMap.put(USER, new LocalMapConfig(10000, 3600));
-        configMap.put(CAPTCHA, new LocalMapConfig(10000, 300));
-        configMap.put(RBAC, new LocalMapConfig(10000, 300));
-
-        configMap.put(URL_MAP, new LocalMapConfig(10000, 300, InMemoryFormat.OBJECT));
-        configMap.put(SCREEN, new LocalMapConfig(1000, 10, InMemoryFormat.OBJECT));
-        configMap.put(VIDEO_TX_ONLINE, new LocalMapConfig(1000, 20, InMemoryFormat.OBJECT));
-        configMap.put(VIDEO_LIVE_MESSAGE, new LocalMapConfig(10000, 86400));
-        configMap.put(VIDEO_LIVE, new LocalMapConfig(10000, 300));
-        configMap.put(VIDEO_LIVE_DELAY, new LocalMapConfig(10000, 10));
-        configMap.put(VIDEO_LIVE_COLUMN, new LocalMapConfig(10000, 300));
-        configMap.put(VIDEO_ACTIVITY, new LocalMapConfig(1000, 300));
-        configMap.put(VIDEO_LIVE_LIBRARY, new LocalMapConfig(1000, 300));
-        configMap.put(VIDEO_LIVE_USER_MAP, new LocalMapConfig(10000, 300));
-        configMap.put(CUSTOMER_MAP, new LocalMapConfig(10000, 3600));
-        configMap.put(QUESTION, new LocalMapConfig(1000, 30));
-
-        configMap.put(COURSE, new LocalMapConfig(10000, 300));
-        configMap.put(GROUP, new LocalMapConfig(10000, 300));
-    }
-
     @Bean
     public Config hazelCastConfig() {
         List<String> memberList = Arrays.asList(members.split(","));
@@ -99,7 +35,7 @@ public class HazelcastConfiguration {
         String instanceName = clusterName + "." + "localIP";
         config.setInstanceName(instanceName);
         config.setClusterName(clusterName);
-        for (Map.Entry<String, LocalMapConfig> entry : configMap.entrySet()) {
+        for (Map.Entry<String, CacheConfig.LocalMapConfig> entry : CacheConfig.getConfigMap().entrySet()) {
             config.addMapConfig(new MapConfig()
                             .setName(entry.getKey())
                             .setEvictionConfig(new EvictionConfig()
@@ -124,7 +60,7 @@ public class HazelcastConfiguration {
 
         // 默认map配置，主要用于USER_VIDEO_TOTAL_ONLINE + videoId
         config.addMapConfig(new MapConfig()
-                .setName(DEFAULT_MAP_NAME)
+                .setName(CacheConfig.DEFAULT_MAP_NAME)
                 .setNearCacheConfig(new NearCacheConfig()
                         .setInMemoryFormat(InMemoryFormat.OBJECT)
                         .setCacheLocalEntries(true)

src/main/java/com/upchina/common/interceptor/WebSocketSessionHandler.java

@@ -7,6 +7,7 @@ import com.upchina.common.entity.OnlineUser;
 import com.upchina.common.vo.FrontUserVO;
 import com.upchina.group.service.common.GroupCacheService;
 import com.upchina.group.service.common.GroupMessageService;
+import com.upchina.video.helper.VideoHelper;
 import com.upchina.video.service.common.VideoCacheService;
 import com.upchina.video.service.common.VideoMessageService;
 import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
@@ -88,8 +89,9 @@ public class WebSocketSessionHandler {
 
     private void updateOnlineStatus(Integer productType, Integer productId, String sessionKey, OnlineUser onlineUser) {
         IMap<String, OnlineUser> totalOnlineMap = getTotalOnlineMap(productType, productId);
+        String cacheKey = VideoHelper.buildOnlineUserCacheKey(productId, sessionKey);
         if (totalOnlineMap != null) {
-            totalOnlineMap.put(sessionKey, onlineUser);
+            totalOnlineMap.put(cacheKey, onlineUser);
         }
     }
 
@@ -103,10 +105,11 @@ public class WebSocketSessionHandler {
         Integer productId = sessionInfo.getProductId();
         String sessionKey = sessionInfo.getSessionKey();
         IMap<String, OnlineUser> totalOnlineMap = getTotalOnlineMap(productType, productId);
-        OnlineUser onlineUser = totalOnlineMap.get(sessionKey);
+        String cacheKey = VideoHelper.buildOnlineUserCacheKey(productId, sessionKey);
+        OnlineUser onlineUser = totalOnlineMap.get(cacheKey);
         if (onlineUser != null) {
             updateOfflineStatus(onlineUser);
-            totalOnlineMap.put(sessionKey, onlineUser);
+            totalOnlineMap.put(cacheKey, onlineUser);
             memberNotify(productType, productId, onlineUser);
         }
     }
@@ -119,9 +122,9 @@ public class WebSocketSessionHandler {
 
     private IMap<String, OnlineUser> getTotalOnlineMap(Integer productType, Integer productId) {
         if (ProductType.VIDEO_SINGLE.value.equals(productType)) {
-            return videoCacheService.getTotalOnlineMap(productId);
+            return videoCacheService.getTotalOnlineMap();
         } else if (ProductType.GROUP.value.equals(productType)) {
-            return groupCacheService.getTotalOnlineMap(productId);
+            return groupCacheService.getTotalOnlineMap();
         }
         return null;
     }

src/main/java/com/upchina/group/service/common/GroupCacheService.java

@@ -4,18 +4,20 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.map.IMap;
+import com.hazelcast.sql.SqlResult;
+import com.hazelcast.sql.SqlRow;
 import com.upchina.advisor.vo.AdvisorBasicVO;
 import com.upchina.common.config.cache.CacheKey;
 import com.upchina.common.constant.IsOrNot;
 import com.upchina.common.entity.OnlineUser;
 import com.upchina.common.service.CacheService;
+import com.upchina.common.util.logger.LoggerUtil;
 import com.upchina.common.vo.DateIdVO;
 import com.upchina.group.constant.GroupInteractiveType;
 import com.upchina.group.constant.GroupMessageUserType;
 import com.upchina.group.constant.QueryGroupMessageType;
 import com.upchina.group.entity.GroupMessage;
 import com.upchina.group.entity.GroupMessageRead;
-import com.upchina.group.entity.GroupUserFlow;
 import com.upchina.group.mapper.GroupMessageMapper;
 import com.upchina.group.mapper.GroupMessageReadMapper;
 import com.upchina.group.mapper.GroupUserFlowMapper;
@@ -26,8 +28,10 @@ import org.springframework.stereotype.Service;
 import javax.annotation.Resource;
 import java.time.LocalDate;
 import java.util.*;
-import java.util.function.Function;
+import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
+
+import static com.upchina.common.config.cache.CacheKey.GROUP_ONLINE_USER;
+import static com.upchina.common.config.cache.CacheKey.VIDEO_LIVE;
 
 @Service
 public class GroupCacheService {
@@ -159,16 +163,80 @@ public class GroupCacheService {
         return vo;
     }
 
-    public IMap<String, OnlineUser> getTotalOnlineMap(Integer groupId) {
+    /**
-        return cacheService.getMap(CacheKey.GroupKey.USER_TOTAL_ONLINE + groupId, () -> {
+     * 获取总在线用户映射
-            synchronized (GroupCacheService.class) {
+     *
-                List<OnlineUser> hisList = groupUserFlowMapper.loadHis(Wrappers.<GroupUserFlow>lambdaQuery()
+     * @return 在线用户映射
-                        .eq(GroupUserFlow::getGroupId, groupId)
+     */
-                        .groupBy(GroupUserFlow::getUserId, GroupUserFlow::getSessionId));
+    public IMap<String, OnlineUser> getTotalOnlineMap() {
-                return hisList.stream()
+        return hazelcastInstance.getMap(GROUP_ONLINE_USER);
-                        .collect(Collectors.toMap(h -> h.getUserId() + "-" + h.getSessionId(), Function.identity()));
     }
-        });
+
+    public List<OnlineUser> getTotalOnlineList(Integer videoId) {
+        String sql = "SELECT * FROM " + GROUP_ONLINE_USER + " WHERE productId = " + videoId;
+        SqlResult result = hazelcastInstance.getSql().execute(sql);
+        List<OnlineUser> list = new ArrayList<>();
+        for (SqlRow row : result) {
+            OnlineUser user = new OnlineUser();
+            user.setProductType(row.getObject("productType"));
+            user.setProductId(row.getObject("productId"));
+            user.setUserId(row.getObject("userId"));
+            user.setUserName(row.getObject("userName"));
+            user.setImg(row.getObject("img"));
+            user.setSessionId(row.getObject("sessionId"));
+            user.setIsOnline(row.getObject("isOnline"));
+            user.setIsPlay(row.getObject("isPlay"));
+            user.setCreateTime(row.getObject("createTime"));
+            user.setExitTime(row.getObject("exitTime"));
+            list.add(user);
+        }
+        return list;
+    }
+
+    public Set<String> getOnlineUserIds(Integer videoId) {
+        String sql = "SELECT distinct userId FROM " + GROUP_ONLINE_USER + " WHERE videoId = " + videoId;
+        SqlResult result = hazelcastInstance.getSql().execute(sql);
+        Set<String> set = new HashSet<>();
+        for (SqlRow row : result) {
+            set.add(row.getObject("userId"));
+        }
+        return set;
+    }
+
+    /**
+     * 获取在线人数
+     *
+     * @param videoId 视频ID
+     * @return 在线人数
+     */
+    public int getOnlineCount(Integer videoId) {
+        IMap<String, Object> map = hazelcastInstance.getMap(VIDEO_LIVE);
+        String cacheKey = CacheKey.VideoLiveKey.ONLINE_COUNT + videoId;
+        Integer onlineCount = (Integer) map.get(cacheKey);
+        if (onlineCount != null) {
+            return onlineCount;
+        }
+        synchronized (this) {
+            onlineCount = (Integer) map.get(cacheKey);
+            if (onlineCount != null) {
+                return onlineCount;
+            }
+            long startTime = System.currentTimeMillis();
+            onlineCount = getOnlineCountWithoutCache(videoId);
+            LoggerUtil.websocket.info("getOnlineCount-" + videoId + ":" + (System.currentTimeMillis() - startTime) + "ms");
+            map.put(cacheKey, onlineCount, 2, TimeUnit.SECONDS);
+        }
+        return onlineCount;
+    }
+
+    public int getOnlineCountWithoutCache(Integer videoId) {
+        String sql = "SELECT COUNT(*) FROM " + GROUP_ONLINE_USER + " WHERE videoId = " + videoId + " and isOnline = 1";
+        SqlResult result = hazelcastInstance.getSql().execute(sql);
+        Iterator<SqlRow> iter = result.iterator();
+        if (iter.hasNext()) {
+            return ((Long)iter.next().getObject(0)).intValue();
+        }
+        return 0;
     }
 
     private static String buildMessageIdSetKey(String prefix, String userId, QueryGroupMessageType type) {

src/main/java/com/upchina/group/service/common/GroupCommonService.java

@@ -105,40 +105,32 @@ public class GroupCommonService {
      */
     @Transactional(rollbackFor = Exception.class)
     public void saveGroupUser() {
-        LocalDateTime now = LocalDateTime.now();
+        IMap<String, OnlineUser> onlineMap = groupCacheService.getTotalOnlineMap();
-        LambdaQueryWrapper<GroupInfo> wrapper = Wrappers.<GroupInfo>lambdaQuery()
+        Collection<OnlineUser> onlineUsers = onlineMap.values();
-                .select(GroupInfo::getId, GroupInfo::getStatus);
+        if (CollUtil.isEmpty(onlineUsers)) {
-        List<GroupInfo> groups = groupInfoMapper.selectList(wrapper);
-        if (CollUtil.isEmpty(groups)) {
             return;
         }
+        LocalDateTime now = LocalDateTime.now();
         LocalDateTime time = now.withSecond(0).withNano(0);
         LocalDateTime startTime = now.plusMinutes(-1);
-
+        List<GroupUserFlow> onLineList = onlineUsers.stream().map(user -> {
-        groups.forEach(g -> {
+            Integer groupId = user.getProductId();
-            Integer id = g.getId();
+            if (IsOrNot.IS.value.equals(user.getIsOnline()) || (user.getExitTime() != null && user.getExitTime().isAfter(startTime))) {
-            IMap<String, OnlineUser> onlineMap = groupCacheService.getTotalOnlineMap(id);
-            if (onlineMap == null || onlineMap.isEmpty()) {
-                return;
-            }
-            // 将实时在线人落库，并添加是否正在观看
-            List<GroupUserFlow> onLineList = onlineMap.values().stream()
-                    .filter(u -> IsOrNot.IS.value.equals(u.getIsOnline()) || (u.getExitTime() != null && u.getExitTime().isAfter(startTime)))
-                    .map(o -> {
                 GroupUserFlow groupUserFlow = new GroupUserFlow();
-                        groupUserFlow.setGroupId(id);
+                groupUserFlow.setGroupId(groupId);
-                        groupUserFlow.setUserId(o.getUserId());
+                groupUserFlow.setUserId(user.getUserId());
-                        groupUserFlow.setSessionId(o.getSessionId());
+                groupUserFlow.setSessionId(user.getSessionId());
                 groupUserFlow.setTime(time);
-                        groupUserFlow.setEnterTime(o.getCreateTime() != null ? o.getCreateTime().withSecond(0).withNano(0) : null);
+                groupUserFlow.setEnterTime(user.getCreateTime() != null ? user.getCreateTime().withSecond(0).withNano(0) : null);
-                        groupUserFlow.setExitTime(o.getExitTime() != null ? o.getExitTime().withSecond(0).withNano(0) : null);
+                groupUserFlow.setExitTime(user.getExitTime() != null ? user.getExitTime().withSecond(0).withNano(0) : null);
                 return groupUserFlow;
-                    })
-                    .collect(Collectors.toList());
-            if (CollUtil.isNotEmpty(onLineList)) {
-                ListUtil.split(onLineList, 1000).stream().filter(CollUtil::isNotEmpty).forEach(groupUserFlowMapper::insertBatchSomeColumn);
             }
-        });
+            return null;
+        }).filter(Objects::nonNull).collect(Collectors.toList());
+        // 将实时在线人落库，并添加是否正在观看
+        if (CollUtil.isNotEmpty(onLineList)) {
+            ListUtil.split(onLineList, 1000).forEach(groupUserFlowMapper::insertBatchSomeColumn);
+        }
     }
 
     public void collectGroupData() {

src/main/java/com/upchina/video/helper/VideoHelper.java

@@ -145,4 +145,8 @@ public class VideoHelper {
         return liveHours;
     }
 
+    public static String buildOnlineUserCacheKey(Integer productId, String sessionKey) {
+        return productId + "-" + sessionKey;
+    }
+
 }

src/main/java/com/upchina/video/service/admin/AdminVideoInteractionService.java

@@ -563,46 +563,43 @@ public class AdminVideoInteractionService {
      */
     @Transactional(rollbackFor = Exception.class)
     public void saveVideoUserDataToDB() {
-        LocalDateTime now = LocalDateTime.now();
+        IMap<String, OnlineUser> onlineMap = videoCacheService.getTotalOnlineMap();
-        LambdaQueryWrapper<VideoLive> wrapper = Wrappers.<VideoLive>lambdaQuery()
+        Collection<OnlineUser> onlineUsers = onlineMap.values();
-                .select(VideoLive::getId, VideoLive::getLiveStatus)
+        if (CollUtil.isEmpty(onlineUsers)) {
-                .ge(VideoLive::getStartTime, now.plusDays(-7));
-        List<VideoLive> videoLives = videoLiveMapper.selectList(wrapper);
-        if (CollUtil.isEmpty(videoLives)) {
             return;
         }
+        Set<Integer> videoIds = onlineMap.values().stream().map(OnlineUser::getProductId).collect(Collectors.toSet());
+        LocalDateTime now = LocalDateTime.now();
+        List<VideoLive> videoLives = videoLiveMapper.selectList(Wrappers.<VideoLive>lambdaQuery().in(VideoLive::getId, videoIds));
+        Map<Integer, Integer> videoStatusMap = videoLives.stream().collect(Collectors.toMap(VideoLive::getId, VideoLive::getLiveStatus));
         LocalDateTime time = now.withSecond(0).withNano(0);
         LocalDateTime startTime = now.plusMinutes(-1);
-
+        List<VideoUserFlow> onLineList = onlineUsers.stream().map(user -> {
-        videoLives.forEach(v -> {
+            Integer videoId = user.getProductId();
-            Integer videoId = v.getId();
+            Integer liveStatus = videoStatusMap.get(videoId);
-            Integer liveStatus = v.getLiveStatus();
+            if (liveStatus == null) {
-            boolean calIsPlay = Objects.equals(VideoLiveStatus.LIVING.value, liveStatus) || Objects.equals(VideoLiveStatus.HAS_ENDED.value, liveStatus);
+                return null;
-            IMap<String, OnlineUser> onlineMap = videoCacheService.getTotalOnlineMap(videoId);
-            if (onlineMap == null || onlineMap.isEmpty()) {
-                return;
             }
-            // 将实时在线人落库，并添加是否正在观看
+            if (IsOrNot.IS.value.equals(user.getIsOnline()) || (user.getExitTime() != null && user.getExitTime().isAfter(startTime))) {
-            List<VideoUserFlow> onLineList = onlineMap.values().stream()
+                boolean calIsPlay = Objects.equals(VideoLiveStatus.LIVING.value, liveStatus) || Objects.equals(VideoLiveStatus.HAS_ENDED.value, liveStatus);
-                    .filter(u -> IsOrNot.IS.value.equals(u.getIsOnline()) || (u.getExitTime() != null && u.getExitTime().isAfter(startTime)))
-                    .map(o -> {
                 VideoUserFlow videoUserFlow = new VideoUserFlow();
                 videoUserFlow.setVideoId(videoId);
-                        videoUserFlow.setUserId(o.getUserId());
+                videoUserFlow.setUserId(user.getUserId());
-                        videoUserFlow.setSessionId(o.getSessionId());
+                videoUserFlow.setSessionId(user.getSessionId());
                 videoUserFlow.setTime(time);
-                        videoUserFlow.setEnterTime(o.getCreateTime() != null ? o.getCreateTime().withSecond(0).withNano(0) : null);
+                videoUserFlow.setEnterTime(user.getCreateTime() != null ? user.getCreateTime().withSecond(0).withNano(0) : null);
-                        videoUserFlow.setExitTime(o.getExitTime() != null ? o.getExitTime().withSecond(0).withNano(0) : null);
+                videoUserFlow.setExitTime(user.getExitTime() != null ? user.getExitTime().withSecond(0).withNano(0) : null);
                 // 当前用户退出的时候，直播处于未开播和暂停状态，则不计入观看时长
                 // 直播中和已结束才计算观看时长
-                        videoUserFlow.setIsPlay(o.getExitTime() != null && calIsPlay ? IsOrNot.IS.value : o.getIsPlay());
+                videoUserFlow.setIsPlay(user.getExitTime() != null && calIsPlay ? IsOrNot.IS.value : user.getIsPlay());
                 return videoUserFlow;
-                    })
+            }
-                    .collect(Collectors.toList());
+            return null;
+        }).filter(Objects::nonNull).collect(Collectors.toList());
+        // 将实时在线人落库，并添加是否正在观看
         if (CollUtil.isNotEmpty(onLineList)) {
             ListUtil.split(onLineList, 1000).forEach(videoUserFlowMapper::insertBatchSomeColumn);
         }
-        });
     }
 
     public static class WatchResultVO {

src/main/java/com/upchina/video/service/admin/AdminVideoStatisticService.java

@@ -15,7 +15,6 @@ import com.upchina.common.config.cache.CacheKey;
 import com.upchina.common.constant.IsOrNot;
 import com.upchina.common.constant.ProductType;
 import com.upchina.common.constant.UserType;
-import com.upchina.common.entity.OnlineUser;
 import com.upchina.common.handler.BizException;
 import com.upchina.common.result.Pager;
 import com.upchina.common.result.ResponseStatus;
@@ -643,8 +642,7 @@ public class AdminVideoStatisticService {
         }
         Page<VideoStatisticUserDetailVO> page = adminVideoInteractionService.selectVideoUserDetail(query.toPage(), wrapper);
         List<VideoStatisticUserDetailVO> records = page.getRecords();
-        List<OnlineUser> onlineList = videoCacheService.getOnlineList(videoId);
+        Set<String> onlineUserIdSet = videoCacheService.getOnlineUserIds(videoId);
-        Set<String> onlineUserIdSet = onlineList.stream().map(OnlineUser::getUserId).collect(Collectors.toSet());
         Map<Integer, String> userNameMap = userService.getUserMap().values().stream().collect(Collectors.toMap(UserDept::getUserId, UserDept::getName));
         Map<String, Integer> readMinuteMap = adminVideoInteractionService.calcuVideoReadMap(videoId, null, null, null);
         Map<String, Integer> messageCountMap = adminVideoInteractionService.calUserMessageCount(videoId, null, VideoMessageContentType.TEXT);
@@ -1054,15 +1052,9 @@ public class AdminVideoStatisticService {
      */
     public UserOnlineVO queryUserOnline(UserOnlineQuery query) {
         Integer videoId = query.getVideoId();
-        List<OnlineUser> onlineList = videoCacheService.getOnlineList(videoId);
+        Set<String> onlineUserIds = videoCacheService.getOnlineUserIds(videoId);
         String userId = query.getUserId();
-        Integer isOnline;
+        Integer isOnline = onlineUserIds.contains(userId) ? IsOrNot.IS.value : IsOrNot.NOT.value;
-        if (onlineList != null && !onlineList.isEmpty()) {
-            OnlineUser entry = onlineList.stream().filter(o -> Objects.equals(o.getUserId(), userId)).findAny().orElse(null);
-            isOnline = entry != null && entry.getIsOnline() != null ? entry.getIsOnline() : IsOrNot.NOT.value;
-        } else {
-            isOnline = IsOrNot.NOT.value;
-        }
         return new UserOnlineVO(userId, videoId, isOnline);
     }
 

src/main/java/com/upchina/video/service/common/VideoCacheService.java

@@ -8,6 +8,8 @@ import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.hazelcast.collection.ISet;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.map.IMap;
+import com.hazelcast.sql.SqlResult;
+import com.hazelcast.sql.SqlRow;
 import com.upchina.advisor.constant.FollowOption;
 import com.upchina.advisor.entity.AdvisorBasic;
 import com.upchina.advisor.service.AdvisorInfoService;
@@ -55,7 +57,6 @@ import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -165,32 +166,41 @@ public class VideoCacheService {
     /**
      * 获取总在线用户映射
      *
-     * @param videoId 视频ID
      * @return 在线用户映射
      */
-    public IMap<String, OnlineUser> getTotalOnlineMap(Integer videoId) {
+    public IMap<String, OnlineUser> getTotalOnlineMap() {
-        return cacheService.getMap(OnlineLineKey.USER_VIDEO_TOTAL_ONLINE + videoId, () -> {
+        return hazelcastInstance.getMap(VIDEO_ONLINE_USER);
-            synchronized (VideoCacheService.class) {
-                List<OnlineUser> hisList = videoUserFlowMapper.loadHis(Wrappers.<VideoUserFlow>query()
-                        .eq("h.video_id", videoId)
-                        .groupBy("h.user_id", "h.session_id"));
-                return hisList.stream()
-                        .collect(Collectors.toMap(h -> h.getUserId() + "-" + h.getSessionId(), Function.identity()));
-            }
-        });
     }
 
-    /**
+    public List<OnlineUser> getTotalOnlineList(Integer videoId) {
-     * 获取在线用户列表
+        String sql = "SELECT * FROM " + VIDEO_ONLINE_USER + " WHERE productId = " + videoId;
-     *
+        SqlResult result = hazelcastInstance.getSql().execute(sql);
-     * @param videoId 视频ID
+        List<OnlineUser> list = new ArrayList<>();
-     * @return 在线用户列表
+        for (SqlRow row : result) {
-     */
+            OnlineUser user = new OnlineUser();
-    public List<OnlineUser> getOnlineList(Integer videoId) {
+            user.setProductType(row.getObject("productType"));
-        IMap<String, OnlineUser> map = getTotalOnlineMap(videoId);
+            user.setProductId(row.getObject("productId"));
-        return map.values().stream()
+            user.setUserId(row.getObject("userId"));
-                .filter(o -> IsOrNot.IS.value.equals(o.getIsOnline()))
+            user.setUserName(row.getObject("userName"));
-                .collect(Collectors.toList());
+            user.setImg(row.getObject("img"));
+            user.setSessionId(row.getObject("sessionId"));
+            user.setIsOnline(row.getObject("isOnline"));
+            user.setIsPlay(row.getObject("isPlay"));
+            user.setCreateTime(row.getObject("createTime"));
+            user.setExitTime(row.getObject("exitTime"));
+            list.add(user);
+        }
+        return list;
+    }
+
+    public Set<String> getOnlineUserIds(Integer videoId) {
+        String sql = "SELECT distinct userId FROM " + VIDEO_ONLINE_USER + " WHERE videoId = " + videoId;
+        SqlResult result = hazelcastInstance.getSql().execute(sql);
+        Set<String> set = new HashSet<>();
+        for (SqlRow row : result) {
+            set.add(row.getObject("userId"));
+        }
+        return set;
     }
 
     /**
@@ -212,13 +222,23 @@ public class VideoCacheService {
                 return onlineCount;
             }
             long startTime = System.currentTimeMillis();
-            onlineCount = getOnlineList(videoId).size();
+            onlineCount = getOnlineCountWithoutCache(videoId);
             LoggerUtil.websocket.info("getOnlineCount-" + videoId + ":" + (System.currentTimeMillis() - startTime) + "ms");
-            map.put(cacheKey, onlineCount, 10, TimeUnit.SECONDS);
+            map.put(cacheKey, onlineCount, 2, TimeUnit.SECONDS);
         }
         return onlineCount;
     }
 
+    public int getOnlineCountWithoutCache(Integer videoId) {
+        String sql = "SELECT COUNT(*) FROM " + VIDEO_ONLINE_USER + " WHERE videoId = " + videoId + " and isOnline = 1";
+        SqlResult result = hazelcastInstance.getSql().execute(sql);
+        Iterator<SqlRow> iter = result.iterator();
+        if (iter.hasNext()) {
+            return ((Long)iter.next().getObject(0)).intValue();
+        }
+        return 0;
+    }
+
     public Integer getVideoSubscribeUserCount(Integer videoId) {
         return getVideoUserSubscribeIds(videoId).size();
     }

src/main/java/com/upchina/video/service/common/VideoCommonService.java

@@ -9,7 +9,6 @@ import com.google.common.collect.Table;
 import com.upchina.common.config.cache.CacheKey;
 import com.upchina.common.constant.IsOrNot;
 import com.upchina.common.constant.ProductType;
-import com.upchina.common.entity.OnlineUser;
 import com.upchina.common.handler.BizException;
 import com.upchina.common.result.ResponseStatus;
 import com.upchina.common.service.*;
@@ -345,8 +344,7 @@ public class VideoCommonService {
         try {
             Integer videoId = video.getId();
             String videoName = video.getTitle();
-            List<OnlineUser> onlineList = videoCacheService.getOnlineList(videoId);
+            Set<String> onlineUserIds = videoCacheService.getOnlineUserIds(videoId);
-            Set<String> onlineUserIds = onlineList.stream().map(OnlineUser::getUserId).collect(Collectors.toSet());
 
             // 查询预约的用户
             LambdaQueryWrapper<VideoLiveUser> subWrapper = Wrappers.<VideoLiveUser>lambdaQuery()

src/main/java/com/upchina/video/service/common/VideoMessageService.java

@@ -1,6 +1,7 @@
 package com.upchina.video.service.common;
 
 import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.exceptions.ExceptionUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONUtil;
 import com.google.common.collect.ImmutableSet;
@@ -24,7 +25,6 @@ import com.upchina.video.vo.cart.CouponVO;
 import com.upchina.video.vo.common.VideoProductInfoVO;
 import com.upchina.video.vo.common.VideoWsMessageVO;
 import com.upchina.video.vo.message.*;
-import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
 import org.springframework.messaging.simp.SimpMessagingTemplate;
 import org.springframework.stereotype.Service;
@@ -34,7 +34,10 @@ import javax.annotation.Resource;
 import java.net.URL;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
-import java.util.*;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 import static com.upchina.common.config.cache.CacheKey.MessageTopicKey;
 
@@ -324,8 +327,8 @@ public class VideoMessageService {
      */
     public void listMember(Integer videoId) {
         //获取进入过直播间的人数
-        IMap<String, OnlineUser> totalOnlineMap = videoCacheService.getTotalOnlineMap(videoId);
+        List<OnlineUser> totalOnlineList = videoCacheService.getTotalOnlineList(videoId);
-        int totalCount = totalOnlineMap.size();
+        int totalCount = totalOnlineList.size();
         LoggerUtil.info("获取直播间在线人列表入参：" + videoId);
         LoggerUtil.info("直播间id=" + videoId + "总人数：" + totalCount);
         VideoCustomerVO videoCustomerVO = new VideoCustomerVO(null, 0, 0);
@@ -333,8 +336,7 @@ public class VideoMessageService {
             //获取在线人数
             int onlineCount = videoCacheService.getOnlineCount(videoId);
             LoggerUtil.info("直播间id=" + videoId + "在线人数：" + onlineCount);
-            List<OnlineUser> onlineUsers = new ArrayList<>(totalOnlineMap.values());
+            videoCustomerVO = new VideoCustomerVO(totalOnlineList, onlineCount, totalCount);
-            videoCustomerVO = new VideoCustomerVO(onlineUsers, onlineCount, totalCount);
         }
         VideoWsMessageVO<VideoCustomerVO> output = VideoWsMessageVO.success(videoId.toString(), videoCustomerVO);
         adminUserTopic.publish(output);
@@ -462,28 +464,29 @@ public class VideoMessageService {
                 LoggerUtil.error("reportPlayStatus上报类型错误, reportType:" + reportType);
                 throw new BizException(ResponseStatus.PARM_ERROR, "上报类型错误");
             }
-            IMap<String, OnlineUser> totalOnlineMap = videoCacheService.getTotalOnlineMap(videoId);
+            IMap<String, OnlineUser> totalOnlineMap = videoCacheService.getTotalOnlineMap();
-            OnlineUser onlineUser = totalOnlineMap.get(sessionKey);
+            String cacheKey = VideoHelper.buildOnlineUserCacheKey(videoId, sessionKey);
+            OnlineUser onlineUser = totalOnlineMap.get(cacheKey);
             if (onlineUser == null) {
-                LoggerUtil.error("reportPlayStatus在线用户不存在, sessionKey:" + sessionKey);
+                LoggerUtil.error("reportPlayStatus在线用户不存在, cacheKey:" + cacheKey);
                 return;
             }
             LoggerUtil.websocket.info("reportPlayStatus上报播放状态, userId:" + userId + ", videoId:" + videoId + ", sessionId:" + sessionId + ", sessionKey:" + sessionKey + ", reportType:" + reportType, ", requestId:" + requestId);
             if (VideoReportType.BACK.value.equals(reportType)) {
                 onlineUser.setIsPlay(IsOrNot.NOT.value);
-                totalOnlineMap.put(sessionKey, onlineUser);
+                totalOnlineMap.put(cacheKey, onlineUser);
             } else if (VideoReportType.FRONT.value.equals(reportType)) {
                 onlineUser.setIsPlay(IsOrNot.IS.value);
-                totalOnlineMap.put(sessionKey, onlineUser);
+                totalOnlineMap.put(cacheKey, onlineUser);
             }
             publishSessionVideoMessage(videoId, sessionId, requestId);
         } catch (BizException e) {
-            LoggerUtil.error("上报播放状态异常:" + ExceptionUtils.getStackTrace(e));
+            LoggerUtil.error("上报播放状态异常:" + ExceptionUtil.stacktraceToString(e));
             if (StrUtil.isNotEmpty(sessionId)) {
                 publishSessionVideoMessage(videoId, sessionId, e);
             }
         } catch (Exception e) {
-            LoggerUtil.error("上报播放状态异常:" + ExceptionUtils.getStackTrace(e));
+            LoggerUtil.error("上报播放状态异常:" + ExceptionUtil.stacktraceToString(e));
             if (StrUtil.isNotEmpty(sessionId)) {
                 publishSessionVideoMessage(videoId, sessionId, new BizException(ResponseStatus.SYS_BUSY, "上报播放状态失败"));
             }