golang实现同一个用户不能同时上传导入压缩包多次的解决办法

345 阅读 0 评论 228 点赞

我是靠谱客的博主名字长了才好记，这篇文章主要介绍golang实现同一个用户不能同时上传导入压缩包多次的解决办法，现在分享给大家，希望可以做个参考。

为了保证同一个用户在上传并处理人脸图片时不能同时进行多次导入操作，可以引入一种分布式锁或者互斥锁机制。这样可以确保同一时间内，同一个用户只能够进行一次上传，后续请求在前一次结束前会被限制。

在您的场景中，我们可以使用几种常见的方式来实现这个功能：

在数据库中添加锁记录：通过在数据库中创建一个字段来标记用户的导入操作是否正在进行中。
使用分布式锁中间件（如 Redis）：通过 Redis 的锁功能来确保同一个用户的并发请求被正确处理。
应用级别的互斥锁：在应用中使用 Go 的 sync.Mutex 来实现，但这种方式更适用于单实例部署，不适合分布式部署场景。

我将介绍每种方法，并详细说明如何实现。

方法 1：在数据库中添加锁记录

可以在数据库中为用户添加一个状态字段，标识当前用户是否正在进行导入操作。当用户开始导入时，设置状态为“处理中”；导入完成后，再清除状态。

代码示例：

首先，在 FaceGroup 表或 User 表中增加一个字段，例如 is_importing：
```
ALTER TABLE FaceGroup ADD COLUMN is_importing BOOLEAN DEFAULT FALSE;
```
在代码中处理状态的检查与设置：

import (
    "context"
    "errors"
    "github.com/gogf/gf/v2/database/gdb"
)

func (s *sFaceGroup) ImportFaceInfo(ctx context.Context, in *v1.FaceInfoImportReq) (res *v1.FaceInfoImportRes, err error) {
    // 检查人脸分组是否存在
    count, err := dao.FaceGroup.Ctx(ctx).Where(dao.FaceGroup.Columns().Id, in.GroupId).Count()
    if err != nil {
        err = cerrors.Wrap(err, consts.InternalServer)
        return
    }
    if count == 0 {
        err = errors.New("当前人脸分组不存在")
        return
    }

    // 检查人脸分组是否正在导入中
    importing, err := dao.FaceGroup.Ctx(ctx).Where(dao.FaceGroup.Columns().Id, in.GroupId).
        Value(dao.FaceGroup.Columns().IsImporting)
    if err != nil {
        err = cerrors.Wrap(err, consts.InternalServer)
        return
    }

    if importing.Bool() {
        err = errors.New("当前分组正在导入中，请稍后再试")
        return
    }

    // 标记当前分组为导入中
    _, err = dao.FaceGroup.Ctx(ctx).Where(dao.FaceGroup.Columns().Id, in.GroupId).
        Data(g.Map{dao.FaceGroup.Columns().IsImporting: true}).Update()
    if err != nil {
        err = cerrors.Wrap(err, consts.InternalServer)
        return
    }

    // 确保在函数返回时清除导入标志
    defer func() {
        _, _ = dao.FaceGroup.Ctx(ctx).Where(dao.FaceGroup.Columns().Id, in.GroupId).
            Data(g.Map{dao.FaceGroup.Columns().IsImporting: false}).Update()
    }()

    // 继续处理导入逻辑...
    // 创建临时目录
    tempDir := gfile.Temp(uuid.New().String())
    defer gfile.Remove(tempDir)
    // 其余代码省略...
}

说明：

在开始导入时，设置 is_importing 为 true，以防止重复导入。
使用 defer 确保无论操作成功还是失败，最终都会清除 is_importing 标志位。
如果检查到 is_importing 为 true，则直接返回错误提示，说明该用户当前已经在导入中。

方法 2：使用 Redis 分布式锁

Redis 是一种高效的分布式锁解决方案，可以用来保证多实例或分布式系统中用户操作的互斥性。

代码示例：

假设您使用 Redis 作为分布式锁的存储，可以引入 Redis 的锁库，比如 redsync。
安装 Redsync：

go get github.com/go-redsync/redsync/v4
go get github.com/go-redis/redis/v8

在代码中使用 Redis 分布式锁：

import (
    "context"
    "errors"
    "time"

    "github.com/go-redsync/redsync/v4"
    "github.com/go-redsync/redsync/v4/redis/goredis/v8"
    "github.com/go-redis/redis/v8"
)

var (
    redisClient = redis.NewClient(&redis.Options{
        Addr: "localhost:6379", // Redis 地址
    })
    pool    = goredis.NewPool(redisClient)
    redsync = redsync.New(pool)
)

func (s *sFaceGroup) ImportFaceInfo(ctx context.Context, in *v1.FaceInfoImportReq) (res *v1.FaceInfoImportRes, err error) {
    // 创建锁
    mutex := redsync.NewMutex("face_group_import_lock_" + in.GroupId,
        redsync.WithExpiry(5*time.Minute), // 锁的过期时间
        redsync.WithTries(1),              // 获取锁的尝试次数
    )

    // 获取锁
    if err = mutex.Lock(); err != nil {
        err = errors.New("当前分组正在导入中，请稍后再试")
        return
    }

    // 确保在函数返回时释放锁
    defer func() {
        if ok, _ := mutex.Unlock(); !ok {
            // 释放锁失败可以进行错误日志记录
            g.Log().Error(ctx, "释放分布式锁失败")
        }
    }()

    // 继续处理导入逻辑...
    // 创建临时目录
    tempDir := gfile.Temp(uuid.New().String())
    defer gfile.Remove(tempDir)
    // 其余代码省略...
}

说明：

使用 redsync 创建分布式锁，可以防止同一个用户在多个实例或并发环境中重复操作。
mutex := redsync.NewMutex("face_group_import_lock_" + in.GroupId) 使用人脸分组 ID 作为锁的键，确保同一分组只能被一个请求处理。
mutex.Lock() 会在获取锁失败时返回错误，这里直接返回提示用户等待。
使用 defer 释放锁，确保无论操作成功或失败都能最终释放锁。

方法 3：应用级别的互斥锁（适用于单实例部署）

对于单实例应用，可以使用 Go 的标准库 sync.Mutex 来实现互斥。对于分布式场景，这种方式不适用。

代码示例：

创建一个全局的 sync.Map 用于存储用户的导入状态，以及 sync.Mutex 进行锁定：

import (
    "context"
    "errors"
    "sync"
)

var (
    importLocks = sync.Map{} // 用于存储用户的锁
)

func (s *sFaceGroup) ImportFaceInfo(ctx context.Context, in *v1.FaceInfoImportReq) (res *v1.FaceInfoImportRes, err error) {
    // 获取或创建锁
    lock, _ := importLocks.LoadOrStore(in.GroupId, &sync.Mutex{})
    userLock := lock.(*sync.Mutex)

    // 加锁
    userLock.Lock()
    defer userLock.Unlock()

    // 继续处理导入逻辑...
    // 创建临时目录
    tempDir := gfile.Temp(uuid.New().String())
    defer gfile.Remove(tempDir)
    // 其余代码省略...
}

说明：