go语言的gorm事务中使用redsync锁锁不住

Question

go语言的gorm事务中使用redsync锁锁不住
在gorm的事务中我开启了20个协程模仿用户删除 但是redsync.Lock()的互斥性消失了 导致都获取了锁
如果我把事务关闭 互斥性就恢复正常了

package main

import (
    "fmt"
    goredislib "github.com/go-redis/redis/v8"
    "github.com/go-redsync/redsync/v4"
    "github.com/go-redsync/redsync/v4/redis/goredis/v8"
    "gorm.io/driver/mysql"
    "gorm.io/gorm"
    "gorm.io/gorm/logger"
    "gorm.io/gorm/schema"
    "log"
    "os"
    "sync"
    "time"
)

var DB *gorm.DB

type BaseModel struct {
    ID        int32          `gorm:"primary_key;comment:ID" json:"id"`
    CreatedAt time.Time      `gorm:"column:add_time;comment:创建时间" json:"-"`
    UpdatedAt time.Time      `gorm:"column:update_time;comment:更新时间" json:"-"`
    DeletedAt gorm.DeletedAt `gorm:"comment:删除时间" json:"-"`
    IsDeleted bool           `gorm:"comment:是否删除" json:"-"`
}

type Inventory struct {
    BaseModel
    Goods   int32 `gorm:"type:int;index;comment:商品id"`
    Stocks  int32 `gorm:"type:int;comment:仓库"`
    Version int32 `gorm:"type:int;comment:分布式锁-乐观锁"`
}

func InitDB() {

    dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=True&loc=Local",
        "root", "123456", "localhost", 3306, "mxshop_inventory_srv2")
    newLogger := logger.New(
        log.New(os.Stdout, "\r\n", log.LstdFlags), // io writer（日志输出的目标，前缀和日志包含的内容——译者注）
        logger.Config{
            SlowThreshold: time.Second, // 慢 SQL 阈值
            LogLevel:      logger.Info, // 日志级别
            //LogLevel: logger.Silent, // 日志级别
            //IgnoreRecordNotFoundError: true,        // 忽略ErrRecordNotFound（记录未找到）错误
            Colorful: true, // 禁用彩色打印
        },
    )
    // 参考 https://github.com/go-sql-driver/mysql#dsn-data-source-name 获取详情
    var err error
    DB, err = gorm.Open(mysql.Open(dsn), &gorm.Config{
        NamingStrategy: schema.NamingStrategy{
            SingularTable: true,
        },
        Logger: newLogger,
    })
    if err != nil {
        panic(err)
    }
}
func main() {
    InitDB()
    client := goredislib.NewClient(&goredislib.Options{
        Addr: "1.1.1.1:6301",
    })
    pool := goredis.NewPool(client)
    rs := redsync.New(pool)
    gNum := 20
    var wg sync.WaitGroup
    wg.Add(gNum)
    DB.Transaction(func(tx *gorm.DB) error {
        for i := 0; i < gNum; i++ {
            go func() {
                defer wg.Done()
                var inv Inventory
                mutex := rs.NewMutex(fmt.Sprintf("goodsss_%d", 421))
                if err := mutex.Lock(); err != nil {
                    fmt.Println("获取redis分布式锁异常-1")
                }
                if result := DB.Where(&Inventory{Goods: int32(421)}).First(&inv); result.RowsAffected == 0 {
                    panic("库存信息不存在")
                }
                fmt.Println(inv.Stocks)
                if err := tx.Model(&Inventory{}).Select("Stocks").Where("goods = ?", int32(421)).Update("stocks", inv.Stocks-1); err.RowsAffected == 0 {
                    fmt.Println("更新失败：", err.Error.Error())
                    fmt.Println(inv.Stocks)
                }
                if ok, err := mutex.Unlock(); !ok || err != nil {
                    fmt.Println("释放redis分布式锁异常-4")
                }
            }()
        }
        return nil
    })
    wg.Wait()
}

Answer 1

这个问题可能是由于在gorm事务中使用redsync锁导致的。因为gorm事务本身已经提供了互斥性保证，因此再使用redsync锁就可能会出现冲突。

解决方案是在事务外部使用redsync锁，并将事务操作放在redsync锁的内部。这样可以确保在同一时刻只有一个操作可以访问数据库，从而避免并发冲突。

补充：如果遇到redis的事务内，执行的修改只有第一条生效的情况，可以尝试以下几种方法来解决：

使用redis的watch命令监控关键字，并在关键字发生改变时重试事务。以下是使用 Redis 的 watch 命令实现事务的示例代码：

import redis

# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 设置关键字的初始值
r.set('key', 100)

while True:
    # 开启事务
    with r.pipeline() as pipe:
        # 监控关键字
        try:
            pipe.watch('key')
            current_value = int(r.get('key'))
            # 对关键字进行操作
            updated_value = current_value - 10
            pipe.multi()
            pipe.set('key', updated_value)
            # 提交事务
            pipe.execute()
            break
        except redis.WatchError:
            # 关键字发生改变，重试事务
            continue

该代码在事务内使用 watch 命令监控关键字，在关键字发生改变时会重试事务。
使用redis的乐观锁机制，在修改前先检查关键字的值是否已经发生改变。

import redis

r = redis.Redis(host='localhost', port=6379, db=0)

def update_value(key, value):
    while True:
        # 获取关键字的当前值
        current_value = r.get(key)
        # 对关键字执行CAS操作，判断关键字的值是否已经发生改变
        if r.getset(key, value) == current_value:
            # 如果关键字的值未发生改变，说明事务执行成功
            return True
        # 否则说明关键字的值已经发生改变，需要重试事务

# 调用update_value函数，更新关键字的值
update_value("key", "new_value")

将多条记录分成多个事务进行处理，避免事务内多条记录产生冲突。
为避免事务内多条记录的冲突，可以将多条记录分成多个事务进行处理。

for record in records:
    redis_conn.multi()
    redis_conn.decrby(record["key"], record["value"])
    redis_conn.exec()

在上面的代码中，对于每条记录，都创建一个独立的事务，在事务内进行扣减操作，这样就可以避免冲突了。

补充2：可以举一个订单系统的例子，假设有多个用户在同时购买同一件商品，如果没有使用事务，那么同时进行扣减库存操作时会出现多次扣减，导致库存减少过多。使用 Redis 事务可以保证在一次事务中，扣减库存操作是原子性的，避免了库存减少过多的情况。但是如果使用的 Redis 事务不是严格的互斥锁，那么多个事务可能同时执行，导致同样的问题。这就需要使用 Redis 分布式锁来避免。如果你想使用事务来保证 Redis 在批量扣减操作中的原子性，可以使用 Redis 事务机制，下面是一个使用 redis-py 库实现的示例代码：

import redis

redis_conn = redis.Redis(host='localhost', port=6379, db=0)

# 开启事务
pipeline = redis_conn.pipeline()
pipeline.multi()

# 执行扣减操作
pipeline.decrby('key1', 10)
pipeline.decrby('key2', 20)

# 执行事务
pipeline.execute()

如果事务在执行过程中遇到任何异常，可以通过捕获异常来回滚事务。

import redis

redis_conn = redis.Redis(host='localhost', port=6379, db=0)

try:
    # 开启事务
    pipeline = redis_conn.pipeline()
    pipeline.multi()

    # 执行扣减操作
    pipeline.decrby('key1', 10)
    pipeline.decrby('key2', 20)

    # 执行事务
    pipeline.execute()
except Exception as e:
    # 回滚事务
    pipeline.discard()
    print("Transaction failed:", e)