实时在线观看人数统计

@Service
@Slf4j
public class OnlineCountService {
    
    private static final String ONLINE_KEY_PREFIX = "video:online:";
    private static final String CLEAN_KEY_PREFIX = "video:clean:";
    private static final int HEARTBEAT_TIMEOUT = 90; // 90秒超时
    private static final int CLEAN_INTERVAL = 60;   // 60秒清理间隔
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    /**
     * 用户进入房间
     */
    public Long enterRoom(String videoId, String userId) {
        String key = ONLINE_KEY_PREFIX + videoId;
        String timestamp = String.valueOf(System.currentTimeMillis() / 1000);
        
        // 原子操作：设置用户在线状态
        redisTemplate.opsForHash().put(key, userId, timestamp);
        
        // 返回当前在线人数
        return getOnlineCount(videoId);
    }
    
    /**
     * 获取在线人数（带清理逻辑）
     */
    public Long getOnlineCount(String videoId) {
        String onlineKey = ONLINE_KEY_PREFIX + videoId;
        String cleanKey = CLEAN_KEY_PREFIX + videoId;
        
        // 尝试设置清理标记
        Boolean needClean = redisTemplate.opsForValue()
            .setIfAbsent(cleanKey, "1", Duration.ofSeconds(CLEAN_INTERVAL));
        
        if (Boolean.TRUE.equals(needClean)) {
            // 异步执行清理
            CompletableFuture.runAsync(() -> cleanExpiredUsers(videoId));
        }
        
        return redisTemplate.opsForHash().size(onlineKey);
    }
    
    /**
     * 清理过期用户
     */
    private void cleanExpiredUsers(String videoId) {
        String key = ONLINE_KEY_PREFIX + videoId;
        long currentTime = System.currentTimeMillis() / 1000;
        
        // 获取所有在线用户
        Map<Object, Object> users = redisTemplate.opsForHash().entries(key);
        
        List<String> expiredUsers = users.entrySet().stream()
            .filter(entry -> {
                long lastActiveTime = Long.parseLong((String) entry.getValue());
                return currentTime - lastActiveTime > HEARTBEAT_TIMEOUT;
            })
            .map(entry -> (String) entry.getKey())
            .collect(Collectors.toList());
        
        // 批量删除过期用户
        if (!expiredUsers.isEmpty()) {
            redisTemplate.opsForHash().delete(key, expiredUsers.toArray());
            log.info("清理视频 {} 的过期用户 {} 个", videoId, expiredUsers.size());
        }
    }
}

// 使用 Pipeline 批量更新心跳
public void batchHeartbeat(Map<String, List<String>> videoUserMap) {
    redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
        long timestamp = System.currentTimeMillis() / 1000;
        
        videoUserMap.forEach((videoId, userIds) -> {
            byte[] key = (ONLINE_KEY_PREFIX + videoId).getBytes();
            userIds.forEach(userId -> {
                connection.hSet(key, userId.getBytes(), 
                    String.valueOf(timestamp).getBytes());
            });
        });
        return null;
    });
}

@Component
public class OnlineCountMetrics {
    
    private final MeterRegistry registry;
    
    // 在线人数指标
    @Scheduled(fixedDelay = 5000)
    public void recordMetrics() {
        videoIds.forEach(videoId -> {
            long count = onlineCountService.getOnlineCount(videoId);
            registry.gauge("video.online.count", 
                Tags.of("video_id", videoId), count);
        });
    }
    
    // 清理耗时监控
    public void recordCleanupTime(String videoId, long duration) {
        registry.timer("video.cleanup.duration",
            Tags.of("video_id", videoId))
            .record(duration, TimeUnit.MILLISECONDS);
    }
}