如何在数据库中实现连续行为streaks的追踪与统计？

触发方案选型对比 #

两种方案没有绝对优劣，按业务规模和需求选择即可：

• 实时触发：行为写入后同步做校验，适合日活在百万级及以下、对徽章发放即时性要求高的场景，逻辑简单好排查问题，唯一的成本是写入链路多了一次状态查询，性能影响可以忽略
• 定时Cron跑批：每日固定时间批量计算全量用户的连续状态，适合超大规模产品、或者统计逻辑非常复杂的streak规则，优点是完全不占写入链路资源，缺点是徽章发放有延迟，需要处理任务重跑的幂等问题

实际生产里很少纯用一种方案，大部分平台都是实时计算做即时反馈，定时任务做兜底校准，兼顾体验和可靠性。

你设计的实时校验方案可行性 #

你提到的带last_date/last_vote_id/last_date_count/total_streak_count字段的表结构逻辑是成立的，只要补两个边界处理就可以直接用：

• 跨天判断不能只校验日期差1天的场景，如果当日和last_date的差值大于1，说明用户中间断了没完成任务，这时候要直接把连续计数重置，不能在原有基础上累加
• 必须加行为唯一ID的幂等校验，避免客户端重试、消息重复投递导致同一次行为被多次计数
  这个方案最大的优势是性能极高：每次投票不需要扫历史投票表，只需要查当前用户的一条streak状态记录即可，单库抗每秒几万次请求毫无压力。

Cron方案的幂等性问题解法 #

任务中断、重跑重复统计的问题解决成本很低，不需要复杂的分布式事务：

• 每次跑批生成一个全局唯一的批次ID，给streak表加last_calc_batch_id字段，计算前先判断当前用户的批次ID是不是和本次批次一致，一致就直接跳过，重跑多少次都不会重复统计
• 计算时不需要拉取用户全量历史行为，只需要取「连续达标要求的最大天数+1」范围内的行为聚合结果即可，比如要求最多连续365天达标，只需要拉最近366天的记录，数据量完全可控。

通用实现伪代码 #

实时触发逻辑 #

// 常量定义：每日达标阈值、对应可获得的徽章规则
DAILY_TARGET_THRESHOLD = 30
ALL_STREAK_BADGES = [{id: "vote_7day", required_streak:7}, {id:"vote_30day", required_streak:30}]

// 用户产生目标行为（投票/访问/发帖等）时触发
function onUserAction(userId, actionId, currentUtcDate):
    // 加用户维度分布式锁，避免并发行为导致计数错误
    lock = acquireLock(f"streak:user:{userId}")
    if not lock:
        return
    
    try:
        // 查询用户当前streak状态，没有则初始化
        streak = getStreakByUserId(userId)
        if not streak:
            streak = {
                "last_date": None,
                "last_action_id": None,
                "current_day_count": 0,
                "total_streak": 0,
                "last_achieved_date": None,
                "badges_awarded": []
            }
        
        // 幂等校验，同一条行为不重复统计
        if streak.last_action_id == actionId:
            return
        
        // 跨天逻辑处理
        if streak.last_date != currentUtcDate:
            dayDiff = calcDateDiff(currentUtcDate, streak.last_date)
            if dayDiff > 1:
                // 断档超过1天，重置连续计数
                streak.total_streak = 0
            // 新的一天，重置当日计数
            streak.current_day_count = 0
            streak.last_date = currentUtcDate
        
        // 当日计数+1
        streak.current_day_count += 1
        streak.last_action_id = actionId
        
        // 判断当日是否达标，达标则累加连续天数
        if streak.current_day_count >= DAILY_TARGET_THRESHOLD and streak.last_achieved_date != currentUtcDate:
            streak.total_streak += 1
            streak.last_achieved_date = currentUtcDate
        
        // 判断是否满足徽章发放条件
        for badge in ALL_STREAK_BADGES:
            if streak.total_streak >= badge.required_streak and badge.id not in streak.badges_awarded:
                sendBadgeNotification(userId, badge.id)
                streak.badges_awarded.append(badge.id)
        
        // 持久化streak状态
        saveStreak(userId, streak)
    finally:
        releaseLock(lock)

定时Cron跑批逻辑 #

// 每日UTC零点后定时触发
function dailyStreakCronJob():
    batchId = generateUniqueBatchId()
    currentUtcDate = getCurrentUtcDate()
    MAX_STREAK_DAYS = max([b.required_streak for b in ALL_STREAK_BADGES])
    startDate = calcDateAdd(currentUtcDate, days= -(MAX_STREAK_DAYS + 1))
    
    // 取最近N+1天的所有用户目标行为，按用户、日期聚合去重
    userDailyActions = aggregateUserActionByDate(startDate, currentUtcDate)
    
    for userActions in userDailyActions:
        userId = userActions.userId
        // 幂等校验，当前批次已经算过的用户跳过
        streak = getStreakByUserId(userId)
        if streak and streak.last_calc_batch_id == batchId:
            continue
        
        // 从当日倒推，计算连续达标天数
        continuousDays = 0
        checkDate = currentUtcDate
        while checkDate >= startDate:
            if userActions.daily_count.get(checkDate, 0) >= DAILY_TARGET_THRESHOLD:
                continuousDays +=1
                checkDate = calcDateAdd(checkDate, days=-1)
            else:
                break
        
        // 初始化streak记录如果不存在
        if not streak:
            streak = {"badges_awarded": []}
        
        // 更新连续天数，校验发放徽章
        streak.total_streak = continuousDays
        streak.last_calc_batch_id = batchId
        for badge in ALL_STREAK_BADGES:
            if streak.total_streak >= badge.required_streak and badge.id not in streak.badges_awarded:
                sendBadgeNotification(userId, badge.id)
                streak.badges_awarded.append(badge.id)
        
        saveStreak(userId, streak)

行业参考实现 #

Stack Overflow的streak类徽章（包括连续访问、连续投票、连续编辑等）用的是实时触发+每日兜底的混合架构：

• 用户产生对应行为时，走实时校验逻辑更新streak状态，满足条件立刻发徽章，保证用户反馈即时
• 每日UTC零点后跑一次全量校准任务，对比用户实际行为数据和streak表的状态，修复因为服务宕机、并发冲突、边界case导致的状态错误，错发的徽章追回、漏发的补发，保证数据最终一致

内容的提问来源于stack exchange，提问作者Lance Pollard