多Pod部署下，如何拒绝同一user_id的并发同步请求？

针对多Pod部署的同步服务，要实现同一user_id的并发Upload请求互斥，本地锁无法跨Pod生效，以下是几种可行的解决方案：

一、数据库行级锁 + 状态字段（推荐优先考虑） #

利用数据库的行级锁机制，结合用户表新增的状态字段来实现跨Pod的请求互斥，不需要额外依赖组件，直接基于现有数据库实现。

实现步骤： #

1. 新增字段：在用户表中添加两个字段：
   • processing_status：枚举类型（如idle/processing），标记用户当前是否有同步请求在处理
   • process_start_time：记录同步请求开始时间，用于处理服务崩溃导致的状态无法重置的情况
2. 请求处理流程：
   • 先尝试将用户的processing_status从idle更新为processing，同时通过FOR UPDATE（MySQL语法，其他数据库对应行锁语法）获取行级锁
   • 如果更新行数为0，说明已有请求在处理，返回409 Conflict；若发现process_start_time超时（比如超过5分钟），可以强制重置状态
   • 处理完数据更新后，将processing_status重置为idle（用defer确保异常时也能执行）

代码示例（Go + MySQL）： #

func (syncServer *SyncServer) Upload(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
    userID := ps.ByName("user_id")
    ctx := r.Context()

    // 1. 尝试获取行锁并更新处理状态
    result, err := syncServer.db.ExecContext(ctx, `
        UPDATE users 
        SET processing_status = 'processing', process_start_time = NOW()
        WHERE id = ? AND processing_status = 'idle'
        FOR UPDATE;
    `, userID)
    if err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        return
    }
    rowsAffected, _ := result.RowsAffected()
    if rowsAffected == 0 {
        // 检查是否是超时的僵死状态
        var processStartTime time.Time
        err := syncServer.db.QueryRowContext(ctx, `
            SELECT process_start_time FROM users WHERE id = ?;
        `, userID).Scan(&processStartTime)
        if err == nil && time.Since(processStartTime) > 5*time.Minute {
            // 超时强制重置状态
            syncServer.db.ExecContext(ctx, `
                UPDATE users SET processing_status = 'idle' WHERE id = ?;
            `, userID)
            w.WriteHeader(http.StatusConflict)
            w.Write([]byte("请求处理超时，请稍后重试"))
        } else {
            w.WriteHeader(http.StatusConflict)
            w.Write([]byte("当前用户已有同步请求正在处理"))
        }
        return
    }

    // 确保无论成功失败都重置状态
    defer func() {
        syncServer.db.ExecContext(ctx, `
            UPDATE users SET processing_status = 'idle' WHERE id = ?;
        `, userID)
    }()

    // 2. 执行数据更新逻辑
    err = syncServer.db.Update(userID, data)
    if err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        return
    }

    w.WriteHeader(http.StatusOK)
}

二、分布式锁（Redis/etcd） #

借助分布式锁组件（如Redis、etcd）实现跨Pod的锁同步，适合对数据库性能敏感的场景。

实现步骤： #

1. 部署分布式锁组件：确保所有Pod都能访问同一个Redis/etcd集群
2. 请求处理流程：
   • 针对每个user_id生成唯一锁键（如sync:lock:{user_id}）
   • 原子性地尝试获取锁，并设置合理的过期时间（防止服务崩溃导致锁永久占用）
   • 获取锁失败则返回409 Conflict；获取成功则处理请求，完成后释放锁

代码示例（Go + Redis）： #

import (
    "context"
    "fmt"
    "net/http"
    "time"

    "github.com/go-redis/redis/v8"
)

func (syncServer *SyncServer) Upload(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
    userID := ps.ByName("user_id")
    lockKey := fmt.Sprintf("sync:lock:%s", userID)
    ctx := r.Context()

    // 尝试获取锁，过期时间5分钟（需大于正常处理时长）
    acquired, err := syncServer.redisClient.SetNX(ctx, lockKey, "processing", 5*time.Minute).Result()
    if err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        return
    }
    if !acquired {
        w.WriteHeader(http.StatusConflict)
        w.Write([]byte("当前用户已有同步请求正在处理"))
        return
    }

    // 延迟释放锁
    defer func() {
        syncServer.redisClient.Del(ctx, lockKey)
    }()

    // 执行数据更新逻辑
    err = syncServer.db.Update(userID, data)
    if err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        return
    }

    w.WriteHeader(http.StatusOK)
}

注意：如果请求处理时长不确定，需要添加锁续约机制（比如定时延长锁的过期时间），避免锁提前过期导致并发请求进入。

三、乐观锁（仅适合数据防覆盖，不适合提前拒绝） #

如果你的需求是防止并发更新导致数据覆盖，而非直接拒绝请求，可以用乐观锁方案，但无法在请求进入处理阶段前拦截。

实现步骤： #

1. 新增字段：在用户表中添加version字段（整数类型，每次更新自增）
2. 请求处理流程：
   • 先查询当前用户的version值
   • 执行更新时带上该version，仅当数据库中的version与查询值一致时才更新
   • 如果更新行数为0，说明有其他请求已更新数据，返回409 Conflict

优缺点对比 #

选择建议 #

• 若已有成熟的数据库集群，优先选数据库行锁+状态字段方案，成本最低
• 若数据库压力较大或对性能要求高，选分布式锁方案
• 若仅需防止数据覆盖而非直接拒绝请求，选乐观锁方案

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