Redis HyperLogLog用户统计功能实现方式代码实用指南

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

# 日级 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)

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"}