Redis定时监控与数据处理实践：从Set到Hash的迁移与优化

SET logscraping:mediaAdId [id1, id2, id3, ...]

HASH logscraping:mediaAdId
    id1 -> { "name": "广告位1", "status": "active" }
    id2 -> { "name": "广告位2", "status": "inactive" }

@Component
@Slf4j
@RequiredArgsConstructor
public class LogMediaAdIdCache {
    private final RedisTemplate<String, String> redisTemplate;
    private volatile Set<Long> cachedLogMediaAdIds = Collections.emptySet();
    public static final String LOG_REDIS_KEY = "logscraping:mediaAdId";

    @PostConstruct
    public void init() {
        refreshCache();
    }

    @Scheduled(fixedRate = 5 * 1000) // 每5秒刷新一次
    public void refreshCache() {
        try {
            Map<Object, Object> idMap = redisTemplate.opsForHash().entries(LOG_REDIS_KEY);
            if (idMap != null) {
                Set<Long> newCache = idMap.keySet().stream()
                        .map(key -> Long.valueOf(key.toString()))
                        .collect(Collectors.toSet());
                this.cachedLogMediaAdIds = newCache;
                log.info("广告位ID缓存刷新成功，数量: {}", newCache.size());
            }
        } catch (Exception e) {
            log.error("刷新广告位ID缓存失败", e);
        }
    }

    public Set<Long> getLogMediaAdIds() {
        return Collections.unmodifiableSet(cachedLogMediaAdIds);
    }

    public boolean contains(Long mediaAdId) {
        return cachedLogMediaAdIds.contains(mediaAdId);
    }

    // 新增方法：获取广告位详情
    public String getAdInfo(Long mediaAdId) {
        return redisTemplate.<String, String>opsForHash().get(LOG_REDIS_KEY, mediaAdId.toString());
    }

    // 新增方法：更新广告位信息
    public void updateAdInfo(Long mediaAdId, String info) {
        redisTemplate.opsForHash().put(LOG_REDIS_KEY, mediaAdId.toString(), info);
        refreshCache(); // 立即刷新缓存
    }
}

@Component
@Slf4j
@RequiredArgsConstructor
public class LogStatsMonitorJob {
    private final RedisTemplate<String, String> redisTemplate;
    private static final String LOG_STATS_KEY = "log:stats:1635474646";

    @Scheduled(fixedRate = 10 * 1000) // 每10秒执行一次
    public void monitorLogStats() {
        String currentTimeKey = getFormattedTime();
        String fullKey = LOG_STATS_KEY + ":" + currentTimeKey;

        try {
            Map<Object, Object> stats = redisTemplate.opsForHash().entries(fullKey);
            if (stats == null || stats.isEmpty()) {
                log.warn("未找到统计信息: {}", fullKey);
                return;
            }

            log.info("----- 统计信息监控（Key: {}）-----", fullKey);
            stats.forEach((field, value) -> 
                log.info("{}: {}", field, value));

            // 业务处理示例
            int total = Integer.parseInt(stats.getOrDefault("total", "0").toString());
            if (total > 1000) {
                log.warn("警告：数据量过大（{}条）", total);
            }
        } catch (Exception e) {
            log.error("监控统计信息失败", e);
        }
    }

    private String getFormattedTime() {
        // 生成格式化的时间戳，如 20250609175050
        return LocalDateTime.now()
                .format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"));
    }
}

@Configuration
@EnableScheduling
public class SchedulingConfig {
    // 启用定时任务
}