Redis 下载量统计：滑动窗口、时间桶与热度排名

Soulock
2026年05月28日
- 技术

两种时间口径

滑动窗口：统计 now() - INTERVAL 到 now() 的下载量，精确反映最近一段时间，但每次查询结果都在变化。

自然时间段：按整点（今天 0 点至今、本小时 0 分至今）统计，实现简单，但整点后计数归零，会出现明显跳变。

大多数业务选择自然时间段作为展示口径，滑动窗口用于后台热度计算。

Redis 分钟级时间桶

每次下载触发 INCR，key 包含资源 ID 和分钟级时间戳：

download:item:123:202605281537

时间戳格式：yyyyMMddHHmm，精度到分钟。

import redis
from datetime import datetime

r = redis.Redis()

def record_download(item_id: int):
    minute_key = datetime.utcnow().strftime("%Y%m%d%H%M")
    key = f"download:item:{item_id}:{minute_key}"
    r.incr(key)
    r.expire(key, 86400 * 2)  # 保留 2 天，超过不再需要

EXPIRE 设为 2 天，避免 key 无限堆积。

查询最近 1 小时下载量

累加过去 60 个分钟 key：

from datetime import datetime, timedelta

def get_downloads_1h(item_id: int) -> int:
    now = datetime.utcnow()
    keys = []
    for i in range(60):
        t = now - timedelta(minutes=i)
        minute_key = t.strftime("%Y%m%d%H%M")
        keys.append(f"download:item:{item_id}:{minute_key}")
    
    counts = r.mget(keys)
    return sum(int(c) for c in counts if c)

同理，3 小时累加 180 个 key，24 小时累加 1440 个 key。

热度排名分数

加权组合多个时间窗口，近期权重更高：

def get_hot_score(item_id: int) -> float:
    h1 = get_downloads_1h(item_id)
    h3 = get_downloads_3h(item_id)
    h24 = get_downloads_24h(item_id)
    return h1 * 0.5 + h3 * 0.3 + h24 * 0.2

定时任务（如每 5 分钟）批量计算所有资源的热度分，写入 Redis Sorted Set：

r.zadd("hot:items", {str(item_id): score})

查询热度榜：

top10 = r.zrevrange("hot:items", 0, 9, withscores=True)

写入性能优化

下载量高时，单个 key 的 INCR 有热点风险。可以用 Pipeline 批量写：

pipe = r.pipeline()
for item_id in download_batch:
    key = f"download:item:{item_id}:{minute_key}"
    pipe.incr(key)
    pipe.expire(key, 172800)
pipe.execute()

或者在应用层做本地聚合，每 30 秒一次性写入 Redis，减少 INCR 频率。

离线统计：ClickHouse

实时热度用 Redis 时间桶，历史下载量分析（按天/按地区）用 ClickHouse：

每次下载写入 Kafka，消费后落表到 ClickHouse
ClickHouse 按 item_id + date 聚合，查询速度远快于 MySQL

Redis 时间桶适合实时排行榜，ClickHouse 适合报表和长周期分析，两者互补。