Redis HyperLogLog用户统计实现代码

一、UV统计的业务诉求与核心指标

用户访问量统计，听起来简单，但背后藏着一个非常实在的问题：怎么准确知道到底有多少人用过你的产品？在互联网产品运营体系中，这直接关系到活跃度、流量规模和产品价值的判断。其中有两个关键指标，几乎每个后端开发都会碰上。

UV（Unique Visitor）独立访客量

简单说，就是一天内访问过网站的自然人数量。同一个用户就算打开十次页面，也只算一次。UV反映的是真实用户覆盖范围，是做用户增长、留存分析的基础。

PV（Page View）页面访问量

用户每点开一个页面，就记一次PV。同一个页面反复刷新，次数累加。PV通常用来衡量网站的总流量与用户粘性——当然，如果PV很高但UV没怎么涨，可能说明用户只是反复点同一个页面，不一定健康。

1.1 传统UV统计方案的局限性

服务端统计UV，核心难点就一个：去重。系统得记住谁来过，谁没来过，而且不能记错。传统方案往往走两条路，但都有明显的坑。

先说关系型数据库。维护一张 user_visit_log 表，记录 (user_id, visit_date) 的唯一索引。逻辑简单，但当用户量上来之后，问题就来了：

• 存储空间爆炸：百万级用户 × 365天 = 3.65亿条记录，索引加数据文件轻松吃掉几十GB磁盘。
• 写入性能劣化：每次访问都要执行 INSERT IGNORE 或 ON DUPLICATE KEY UPDATE，B+树频繁分裂合并，数据库CPU和IO很快就扛不住了。
• 查询延迟飙升：统计某日UV时，SELECT COUNT(DISTINCT user_id) ... 必然触发全表扫描或临时表排序，数据量一大，响应时间直接指数级上升。

那换Redis Set呢？内存消耗一样让人头疼。假设日活用户1000万，每个用户ID占8字节（Long类型），单日就得80MB内存。全年累计下来29GB，成本高昂，而且不可持续。

二、Redis HyperLogLog的核心优势

2.1 算法原理与核心特性

HyperLogLog（HLL）是一种概率算法，专门用来估算大量数据的基数（即不同元素的数量），而且不需要存储所有值。它的核心思路非常巧妙：利用哈希函数把元素均匀打散，然后观察哈希值的二进制模式中前导零的个数，从中推断出大概有多少个不同的元素。

Redis里的HyperLogLog基于String结构实现，单个HLL的内存占用永远小于16KB。这个压缩比简直逆天——用12KB就能代替80MB，省了99.99%内存，这才是真正的内存优化。

代价与权衡

既然是概率算法，就有误差。HyperLogLog的标准误差小于 0.81%。但在UV统计这类场景里，这个误差完全可以忽略——运营同学更关心流量趋势和量级，而不是精确到个位数的数字。

2.2 Redis HyperLogLog核心命令详解

命令	功能描述	时间复杂度
PFADD key element [element ...]	添加一个或多个元素到HyperLogLog	O(1)
PFCOUNT key [key ...]	计算一个或多个HyperLogLog的并集基数	O(N)
PFMERGE destkey sourcekey [sourcekey ...]	将多个HyperLogLog合并为一个	O(N)

命令详解：

• PFADD：向指定Key的HyperLogLog中添加元素。如果元素已经存在，不会重复计数。返回值1表示HyperLogLog被修改（新增元素），0表示元素已存在。
• PFCOUNT：返回指定Key的估算基数。支持传入多个Key，返回它们的并集去重数量，非常适合跨天、跨维度的合并统计。
• PFMERGE：将多个源HyperLogLog合并到目标Key中，适合数据归档与离线分析。

三、UV统计实现与压测验证

3.1 单元测试压测代码与结果

我们通过一段单元测试向HyperLogLog中添加100万条数据，看看它到底能省多少内存，精度怎么样：

import redis
import time

def test_hyperloglog():
    # 连接 Redis
    r = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)
    # 清空测试 Key
    r.delete('hll:uv:daily')
    # 准备批量添加
    batch_size = 1000
    total_users = 1000000
    start_time = time.time()
    # 批量添加 100 万用户
    for i in range(0, total_users, batch_size):
        users = [f"user_{j}" for j in range(i + 1, min(i + batch_size + 1, total_users + 1))]
        r.pfadd('hll:uv:daily', *users)
    # 统计数量
    uv_count = r.pfcount('hll:uv:daily')
    # 获取内存占用
    memory_used = r.memory_usage('hll:uv:daily')
    elapsed_time = time.time() - start_time
    print(f"实际添加用户数: {total_users}")
    print(f"HyperLogLog 统计结果: {uv_count}")
    print(f"误差率: {abs(uv_count - total_users) / total_users * 100:.4f}%")
    print(f"内存占用: {memory_used} bytes ({memory_used / 1024:.2f} KB)")
    print(f"耗时: {elapsed_time:.4f} 秒")
    print(f"吞吐量: {total_users / elapsed_time:.0f} ops/sec")

if __name__ == "__main__":
    test_hyperloglog()

压测结果分析（典型输出）：

实际添加用户数: 1000000
HyperLogLog 统计结果: 1008542
误差率: 0.8542%
内存占用: 12288 bytes (12.00 KB)
耗时: 2.3456 秒
吞吐量: 426315 ops/sec

结论：

• 内存占用：仅12KB，远低于Set结构的80MB（100万用户）。
• 统计精度：误差率0.85%，符合HyperLogLog的标准误差范围（< 0.81%为理论值，实际略有波动）。
• 写入性能：每秒可处理42万次添加操作，完全满足高并发场景。

3.2 UV统计架构设计与实现

Key设计规范：

# 日级 UV 统计
uv:daily:{YYYY-MM-DD}  # 例如: uv:daily:2024-01-15
# 月级 UV 统计（通过 PFCOUNT 合并日级数据）
uv:monthly:{YYYY-MM}    # 例如: uv:monthly:2024-01
# 全站历史 UV（通过 PFMERGE 合并月级数据）
uv:lifetime

核心实现代码：

import redis
from datetime import datetime, timedelta
from typing import List

class UVStatisticsService:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def record_visit(self, user_id: int, visit_date: datetime = None):
        """
        记录用户访问
        :param user_id: 用户 ID
        :param visit_date: 访问日期（默认当天）
        """
        if visit_date is None:
            visit_date = datetime.now()
        date_key = visit_date.strftime("%Y-%m-%d")
        key = f"uv:daily:{date_key}"
        # 添加用户到当日 HyperLogLog
        self.redis.pfadd(key, str(user_id))
        # 设置过期时间（保留 90 天数据）
        self.redis.expire(key, 90 * 24 * 60 * 60)

    def get_daily_uv(self, date: datetime = None) -> int:
        """
        获取指定日期的 UV
        :param date: 查询日期（默认当天）
        :return: UV 数量
        """
        if date is None:
            date = datetime.now()
        date_key = date.strftime("%Y-%m-%d")
        key = f"uv:daily:{date_key}"
        return self.redis.pfcount(key)

    def get_date_range_uv(self, start_date: datetime, end_date: datetime) -> int:
        """
        获取日期范围内的 UV（并集统计）
        :param start_date: 开始日期
        :param end_date: 结束日期
        :return: 去重后的 UV 数量
        """
        keys = []
        current_date = start_date
        while current_date <= end_date:
            key = f"uv:daily:{current_date.strftime('%Y-%m-%d')}"
            keys.append(key)
            current_date += timedelta(days=1)
        if not keys:
            return 0
        # 使用 PFCOUNT 计算并集基数
        return self.redis.pfcount(*keys)

    def merge_monthly_uv(self, year: int, month: int):
        """
        合并月度 UV 统计
        :param year: 年份
        :param month: 月份
        """
        # 生成该月所有日期的 Key
        keys = []
        date = datetime(year, month, 1)
        while date.month == month:
            key = f"uv:daily:{date.strftime('%Y-%m-%d')}"
            keys.append(key)
            date += timedelta(days=1)
        # 合并到月度 Key
        monthly_key = f"uv:monthly:{year}-{month:02d}"
        self.redis.pfmerge(monthly_key, *keys)
        # 设置过期时间（保留 2 年）
        self.redis.expire(monthly_key, 2 * 365 * 24 * 60 * 60)

FastAPI接口实现：

from fastapi import APIRouter, Depends, HTTPException
from datetime import datetime, timedelta
from pydantic import BaseModel

router = APIRouter()

class UVStatsResponse(BaseModel):
    daily_uv: int
    weekly_uv: int
    monthly_uv: int

@router.get("/stats/uv", response_model=UVStatsResponse)
async def get_uv_statistics(
    date: str = None,
    uv_service: UVStatisticsService = Depends(lambda: UVStatisticsService(redis_client))
):
    """
    获取 UV 统计数据
    :param date: 查询日期（YYYY-MM-DD 格式，默认当天）
    :return: 日、周、月 UV 统计
    """
    if date:
        try:
            query_date = datetime.strptime(date, "%Y-%m-%d")
        except ValueError:
            raise HTTPException(status_code=400, detail="Invalid date format")
    else:
        query_date = datetime.now()
    # 日 UV
    daily_uv = uv_service.get_daily_uv(query_date)
    # 周 UV（最近 7 天）
    week_start = query_date - timedelta(days=6)
    weekly_uv = uv_service.get_date_range_uv(week_start, query_date)
    # 月 UV（最近 30 天）
    month_start = query_date - timedelta(days=29)
    monthly_uv = uv_service.get_date_range_uv(month_start, query_date)
    return UVStatsResponse(
        daily_uv=daily_uv,
        weekly_uv=weekly_uv,
        monthly_uv=monthly_uv
    )

@router.post("/visit/record")
async def record_user_visit(
    user_id: int,
    uv_service: UVStatisticsService = Depends(lambda: UVStatisticsService(redis_client))
):
    """
    记录用户访问
    :param user_id: 用户 ID
    """
    uv_service.record_visit(user_id)
    return {"status": "success", "message": "Visit recorded"}

四、HyperLogLog与Bitmap、Set的对比选型

数据结构	内存占用	精确度	适用场景	核心命令
HyperLogLog	< 16KB（固定）	误差 < 0.81%	海量UV统计、去重计数	PFADD, PFCOUNT
Bitmap	N bits（N=最大值）	100%精确	连续整数ID签到、状态标记	SETBIT, GETBIT, BITCOUNT
Set	N × 8 bytes	100%精确	中小规模去重集合、交集/并集运算	SADD, SISMEMBER, SINTER

选型建议：

• UV统计（千万级以上）：HyperLogLog，内存占用极低，误差可接受。
• 用户签到（连续日期）：Bitmap，按日期偏移量存储，支持连续签到统计。
• 共同关注/好友列表（万级以下）：Set，支持交集、并集等集合运算，精确度高。

五、总结与优化建议

5.1 核心收益

1. 极致内存效率：单日UV统计仅需12KB内存，全年365天累计仅4.5MB，相比Set结构节省99.99%内存。
2. 高性能写入：单机Redis可支撑百万级QPS的PV记录，满足亿级日活产品的统计需求。
3. 灵活聚合能力：通过 PFCOUNT 与 PFMERGE 实现跨天、跨维度的并集统计，支持周报、月报等复杂分析场景。

5.2 优化建议

1. TTL过期策略：为日级UV Key设置90天TTL，自动清理历史数据，防止内存无限增长。
2. 月度归档：通过定时任务执行 PFMERGE，将日级数据合并为月度Key，便于长期趋势分析。
3. 误差容忍：业务层需明确HyperLogLog的误差特性，避免在财务结算等强一致性场景使用。
4. 监控告警：监控Redis内存使用率与HyperLogLog Key数量，设置阈值告警，防止内存溢出。