本文基于「食友」——一个校园美食点评平台的后端项目，分享 Redis 在四个业务场景中的落地方式：评价互动、互动落库、排行榜刷新和后台权限校验。

背景

食友是一个面向大学生的校园美食点评平台，类似「校园版大众点评」。用户可以对校内外的店铺和菜品发布多维度评价（口味、价格、环境、服务），并对评价进行点赞/踩互动。

技术栈：Go + Gin + GORM + Redis + MySQL。

在项目里，Redis 不只是普通缓存，而是同时承担实时状态、异步削峰、并发控制和高频权限判断四类职责：

1. 评价互动：用 Lua 保证点赞/踩原子更新
2. 互动落库：用 Streams 做异步队列
3. 排行榜刷新：用分布式锁控制并发
4. 权限校验：用 Set 缓存管理员权限集合

下面逐一展开。

---

一、评价互动：Lua 保证点赞/踩原子更新

问题

用户对一条评价可以点赞（1）、踩（0）或取消。这三种操作需要原子地完成：

• 首次操作：写入用户动作，对应计数器 +1
• 重复点击相同动作：取消，删除用户动作，计数器 -1
• 切换动作（赞→踩 或 踩→赞）：更新用户动作，两个计数器一增一减

如果用普通的 GET + 判断 + SET 多步操作，在并发场景下会出现竞态条件——两个请求同时读到旧状态，导致计数器不一致。

方案

用 Redis Lua 脚本，一次 EVAL 调用完成全部逻辑：

local statsKey = KEYS[1]      -- review:stats:{reviewID}  Hash 存计数
local reactionKey = KEYS[2]   -- review:reaction:{reviewID}:{userID}  存用户动作
local newAction = tonumber(ARGV[1]) -- 1 表示点赞，0 表示踩

-- 先读取用户之前的动作，用它决定是新增、取消还是切换
local oldAction = redis.call("GET", reactionKey)

-- 首次操作
if oldAction == false then
    redis.call("SET", reactionKey, newAction)
    if newAction == 1 then
        redis.call("HINCRBY", statsKey, "like_count", 1)
    else
        redis.call("HINCRBY", statsKey, "dislike_count", 1)
    end
    return {1, newAction} -- 1 表示新增动作

-- 重复点击相同动作（取消）
elseif oldAction == tostring(newAction) then
    redis.call("DEL", reactionKey)
    if newAction == 1 then
        redis.call("HINCRBY", statsKey, "like_count", -1)
    else
        redis.call("HINCRBY", statsKey, "dislike_count", -1)
    end
    return {2, 0} -- 2 表示取消，当前动作归零

-- 切换操作（赞-踩 或 踩-赞）
else
    redis.call("SET", reactionKey, newAction)
    if newAction == 1 then
        redis.call("HINCRBY", statsKey, "like_count", 1)
        redis.call("HINCRBY", statsKey, "dislike_count", -1)
    else
        redis.call("HINCRBY", statsKey, "dislike_count", 1)
        redis.call("HINCRBY", statsKey, "like_count", -1)
    end
    return {3, newAction} -- 3 表示从赞切到踩，或从踩切到赞
end

Go 侧调用：

func (c *interactionCache) ToggleReactionRedis(ctx context.Context, reviewID, userID int, actionType int8) (int, int8, error) {
    // 计数和用户动作拆成两个 key：一个存聚合结果，一个存当前用户状态。
    statsKey := fmt.Sprintf("review:stats:%d", reviewID)
    reactionKey := fmt.Sprintf("review:reaction:%d:%d", reviewID, userID)

    // Lua 脚本在 Redis 内一次执行完，避免应用层多次 GET/SET 的竞态。
    res, err := c.rdb.Eval(ctx, toggleReactionLua, []string{statsKey, reactionKey}, actionType).Result()
    if err != nil {
        return 0, 0, fmt.Errorf("redis lua failed: %w", err)
    }

    slice, ok := res.([]interface{})
    if !ok || len(slice) < 2 {
        return 0, 0, fmt.Errorf("unexpected lua result")
    }
    // changed 表示操作类型，currentAction 表示最终状态，直接返回给业务层更新 UI。
    changed, _ := strconv.Atoi(fmt.Sprint(slice[0]))
    currentAction, _ := strconv.Atoi(fmt.Sprint(slice[1]))
    return changed, int8(currentAction), nil
}

要点

• Redis 执行 Lua 脚本是单线程原子的，不需要额外加锁
• 返回值 {changed, currentAction} 让调用方知道操作类型和当前状态，便于前端即时更新 UI
• 计数用 Hash（HINCRBY），用户动作用 String（GET/SET/DEL），数据结构选择贴合业务

---

二、互动落库：Redis Streams 做异步队列

问题

点赞/踩操作写入 Redis 后需要异步同步到 MySQL。直接在请求链路里写库会拖慢响应，而且高并发下数据库压力大。

方案

用 Redis Streams 实现生产者-消费者模型：

用户请求 → Lua 原子操作(Redis) → 发布到 Stream → Consumer 消费 → 批量写入 MySQL

生产者（写完 Redis 后发消息）：

func (c *interactionCache) SyncReactionToDB(ctx context.Context, reviewID, userID int, actionType int8) {
    // Redis 已经完成实时状态更新，这里只投递最终需要异步落库的事件。
    msg := &mq.ReactionMessage{
        ReviewID:   reviewID,
        UserID:     userID,
        ActionType: actionType,
        Timestamp:  time.Now().Unix(),
    }
    // 投递失败只记日志，不影响本次互动的实时返回。
    if err := c.redisStream.PublishReaction(ctx, msg); err != nil {
        logger.Warn("[InteractionCache] publish reaction to queue failed", zap.Error(err))
    }
}

消费者（批量消费 + 消息合并 + 事务写库）：

func (c *InteractionConsumer) consumeLoop() {
    const (
        batchSize     = 100             // 单批最多写库 100 条
        blockDuration = 2 * time.Second // 没消息时最多阻塞等待 2 秒
        flushInterval = 3 * time.Second // 即使没满 100 条，也定时刷一次
    )

    batch := make([]mq.ReactionMessage, 0, batchSize)
    ticker := time.NewTicker(flushInterval)
    defer ticker.Stop()

    for {
        select {
        case <-c.stopCh:
            // 服务退出前把内存中的剩余消息刷掉，减少丢数据窗口。
            if len(batch) > 0 {
                c.flushBatch(context.Background(), batch, messageIDs)
            }
            return
        case <-ticker.C:
            // 用时间阈值控制延迟，避免低流量时消息一直攒在内存里。
            if len(batch) > 0 {
                c.flushBatch(context.Background(), batch, messageIDs)
                batch = batch[:0]
            }
        default:
            messages, ids, err := c.redisStream.Consume(ctx, batchSize, blockDuration)
            if len(messages) > 0 {
                batch = append(batch, messages...)
                // 用数量阈值控制吞吐，攒够一批就写库。
                if len(batch) >= batchSize {
                    c.flushBatch(context.Background(), batch, messageIDs)
                    batch = batch[:0]
                }
            }
        }
    }
}

消息合并

这是整个方案的亮点。同一个用户对同一条评价的多次操作会被合并成最终结果：

// 消息合并：同一个 reviewID+userID 只保留最终操作
reactionOps := make(map[string]reactionOp)
for _, msg := range messages {
    // 用户和评价组成唯一维度，同一维度内只保留最后一次操作。
    key := fmt.Sprintf("%d:%d", msg.ReviewID, msg.UserID)
    if existing, ok := reactionOps[key]; ok {
        if msg.ActionType == 0 {
            // actionType 为 0 时表示取消，最终落库应删除互动记录。
            reactionOps[key] = reactionOp{..., del: true}
        } else if existing.actionType != 0 && existing.actionType != msg.ActionType {
            // 从赞切到踩，或从踩切到赞，只保留新的动作。
            reactionOps[key] = reactionOp{..., actionType: msg.ActionType}
        }
    } else {
        // 第一次遇到该用户对该评价的操作，先放入合并表。
        reactionOps[key] = reactionOp{..., actionType: msg.ActionType}
    }
}

举个例子：用户 A 先点赞、再踩、再取消，产生 3 条消息。合并后只需 1 条 DELETE 操作写库。写放大问题直接消解。

要点

• 用 Consumer Group 保证消息不重复消费
• XACK 在写库成功后才确认，保证 at-least-once 语义
• 3 秒定时刷新 + 100 条批量上限，平衡延迟和吞吐
• Stream 设置 MaxLen: 100000 + Approx: true，自动裁剪旧消息，防止内存无限增长

---

三、排行榜刷新：分布式锁控制并发

问题

排行榜通过定时任务（cron）每 30 分钟刷新一次，也支持手动触发。如果多个实例同时刷新，会产生数据冲突。

方案

用 Redis SETNX + Lua 释放锁实现分布式锁：

// 加锁：SETNX + TTL
func (s *rankingService) tryLock(ctx context.Context, value string) (bool, error) {
    // key 不存在时才写入成功，同时设置 TTL 防止进程异常退出后锁永久存在。
    return s.rdb.SetNX(ctx, rankingRefreshLockKey, value, rankingRefreshLockTTL).Result()
}

// 解锁：Lua 脚本保证只有持锁者才能释放
func (s *rankingService) unlock(ctx context.Context, value string) {
    const script = `
-- 只允许 value 匹配的持锁者删除锁，避免误删其他实例刚拿到的锁
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
end
return 0
`
    _ = s.rdb.Eval(ctx, script, []string{rankingRefreshLockKey}, value).Err()
}

使用方式：

// value 必须能唯一标识当前实例的本次刷新任务。
lockValue := fmt.Sprintf("instance-%d-%d", os.Getpid(), time.Now().UnixNano())
acquired, err := s.tryLock(ctx, lockValue)
if !acquired {
    return fmt.Errorf("排行榜正在刷新中，请稍后重试")
}
// defer 保证正常返回或中途出错时都会尝试释放锁。
defer s.unlock(ctx, lockValue)

// 执行刷新逻辑...

要点

• value 用 PID + 时间戳唯一标识当前持锁者，防止误释放其他实例的锁
• 释放锁时先 GET 比较再 DEL，用 Lua 保证原子性——这是经典的「check-and-delete」模式
• TTL（30 分钟）作为安全网，即使进程崩溃锁也会自动释放
• 锁的粒度是「全站排行榜刷新」，而不是单个榜单，减少锁竞争

---

四、权限校验：Set 缓存管理员权限集合

问题

后台接口不是只校验“是否登录”，还要校验当前管理员是否有访问某个 API 的权限。权限判断发生在 VerifyAPIPermission 中间件里，访问后台接口时会频繁触发。

如果每次都根据管理员、角色、角色权限、权限表去 MySQL 做关联查询，请求量上来后会产生两个问题：

• 高频后台接口会反复打数据库
• 无权限或空权限的管理员也会重复回源查询

权限数据本身读多写少，适合放进 Redis 做读缓存。

方案

项目里把“管理员拥有的一组权限”缓存成 Redis Set。实现位于 internal/repository/admin_permission_cache_repository.go。

Key 设计是“管理员账号 -> 权限集合”：

• Key 前缀：admin:permission:user:
• 完整 Key：admin:permission:user:{adminUserID}
• Value 结构：Redis Set（成员是权限 key）

例如：

• admin:permission:user:1001
• Set members: shop:list、shop:update、report:review

这个结构天然适合权限判断：校验某个权限是否存在时，直接 SISMEMBER，时间复杂度稳定。

const (
    adminPermissionCachePrefix = "admin:permission:user:"
    emptyPermissionMarker      = "__empty__" // 区分“缓存不存在”和“确实没有权限”
)

func (r *adminPermissionCacheRepository) HasUserPermission(ctx context.Context, adminUserID int, permissionKey string) (bool, bool, error) {
    key := adminPermissionCacheKey(adminUserID)

    // 先判断 key 是否存在：不存在表示缓存 miss，需要回源 DB。
    exists, err := r.rdb.Exists(ctx, key).Result()
    if err != nil {
        return false, false, err
    }
    if exists == 0 {
        return false, false, nil // 第一个 false 表示未授权，第二个 false 表示未命中缓存
    }

    // key 存在后再判断权限集合里是否有目标 permissionKey。
    allowed, err := r.rdb.SIsMember(ctx, key, permissionKey).Result()
    return allowed, true, err // 第二个 true 表示 Redis 已命中
}

写入缓存时，如果管理员没有任何权限，也不会让 Redis key 缺失，而是写入一个空权限标记：

代码里定义了：emptyPermissionMarker = "__empty__"。

目的不是“加一个假权限”，而是明确区分两种状态：

1. Redis 里压根没有这个 key（真正 miss）
2. Redis key 存在，但该管理员确实没有任何权限

如果不做这个区分，第二种情况会被误判为 miss，导致每次都回源数据库，形成无效穿透。

读路径

1. 先查 Redis：EXISTS + SISMEMBER
2. 如果 key 不存在（miss），回源 DB 查询该管理员权限集合
3. 将结果写回 Redis Set（空集合则写入 __empty__）
4. 设置 TTL，后续请求直接走缓存

这一步“miss 后立即回写”本质是在做 cache-aside：

• 对热点管理员权限，避免重复打 DB（防击穿）
• 对本来就无权限的管理员，避免反复空查（借助 __empty__ 防穿透）

失效策略

项目里在角色/权限变更时会删除对应管理员权限缓存（以及部分场景全量清理）：

• DeleteUserPermissions([]adminUserIDs)：精准删
• ClearUserPermissions()：批量场景下全量删

这保证了权限变更后，下一次请求会 miss 并回源，拿到新权限再回写缓存。

要点

• 权限集合用 Redis Set 表达最贴合，SISMEMBER 判断开销低
• miss 后回写属于典型 cache-aside，可降低热点权限查询的数据库压力
• __empty__ 空标记可避免“无权限管理员”反复回源，防止穿透
• 配合权限变更时的删除缓存，读链路性能和正确性可以兼顾

---

总结

这篇文章一共分成四种 Redis 实战模式：

业务场景	Redis 技术	关键点
评价互动	Lua 脚本	点赞/踩/取消/切换在 Redis 中原子完成
互动落库	Streams + Consumer Group	异步削峰，消息合并减少写放大
排行榜刷新	SETNX + Lua 释放锁	多实例下只允许一个刷新任务执行
权限校验	Set + cache-aside	SISMEMBER 快速判断，miss 后回写，空标记防穿透

Redis 不只是缓存。用好 Lua 脚本、Streams、分布式锁和 Set 缓存这些能力，可以在不引入 Kafka/RabbitMQ 等重依赖的情况下，解决很多实际工程问题。对于中小规模项目，这种方案在开发效率和运维成本上都更轻。