Mysql数据库乐观锁与悲观锁示例详解

-- 排他锁（写锁）：其他事务不能读（除非快照读）、不能写
SELECT * FROM accounts WHERE id = 1 FOR UPDATE;
-- 共享锁（读锁）：允许多个事务读，但阻止写
SELECT * FROM accounts WHERE id = 1 FOR SHARE;

func TransferHandler(c *gin.Context) {
    tx := db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx.Rollback()
        }
    }()
    var fromAccount Account
    // 悲观锁：锁定 from 账户
    if err := tx.Set("gorm:query_option", "FOR UPDATE").
        Where("id = ?", 1).First(&fromAccount).Error; err != nil {
        tx.Rollback()
        c.JSON(400, gin.H{"error": "账户不存在"})
        return
    }
    var toAccount Account
    if err := tx.Set("gorm:query_option", "FOR UPDATE").
        Where("id = ?", 2).First(&toAccount).Error; err != nil {
        tx.Rollback()
        c.JSON(400, gin.H{"error": "目标账户不存在"})
        return
    }
    if fromAccount.Balance < 100 {
        tx.Rollback()
        c.JSON(400, gin.H{"error": "余额不足"})
        return
    }
    fromAccount.Balance -= 100
    toAccount.Balance += 100
    tx.Save(&fromAccount)
    tx.Save(&toAccount)
    tx.Commit()
    c.JSON(200, gin.H{"msg": "转账成功"})
}

CREATE TABLE products (
    id INT PRIMARY KEY,
    name VARCHAR(100),
    stock INT,
    version INT DEFAULT 0
);

UPDATE products 
SET stock = stock - 1, version = version + 1 
WHERE id = 1 AND version = 5;  -- 只有 version 未变才更新

type Product struct {
    ID      uint `gorm:"primarykey"`
    Name    string
    Stock   int
    Version uint32 // GORM 自动识别为乐观锁字段
}
func ReduceStock(c *gin.Context) {
    var product Product
    id := c.Param("id")
    // 第一次读取
    if err := db.First(&product, id).Error != nil {
        c.JSON(404, gin.H{"error": "商品不存在"})
        return
    }
    // 业务逻辑：扣减库存
    if product.Stock <= 0 {
        c.JSON(400, gin.H{"error": "库存不足"})
        return
    }
    // 尝试更新（GORM 自动在 WHERE 中加入 version 条件）
    product.Stock--
    result := db.Save(&product)
    if result.Error != nil {
        c.JSON(500, gin.H{"error": "数据库错误"})
        return
    }
    if result.RowsAffected == 0 {
        // 乐观锁失败：版本不匹配
        c.JSON(409, gin.H{"error": "库存已被其他请求修改，请重试"})
        return
    }
    c.JSON(200, gin.H{"msg": "扣减成功", "stock": product.Stock})
}

用户请求
    │
    ▼
[ Gin Web 服务 ] ←─┐
    │              │
    ▼              │
[ Redis 预减库存 ] │ ←─ 库存校验 & 原子扣减（Lua 脚本）
    │              │
    ▼              │
[ 发送消息到 MQ ] ─┘ → [ Kafka / RabbitMQ / RocketMQ ]
    │
    ▼
[ 异步消费服务 ]
    │
    ▼
[ MySQL 落库 ] ←─ 订单创建、库存最终扣减、记录日志
    │
    ▼
[ 返回结果给用户（可延迟）]

// 初始化库存（管理后台或定时任务调用）
redisClient.Set(ctx, "stock:product:1001", 100, 0)

-- stock_decrease.lua
local key = KEYS[1]
local userId = ARGV[1]
-- 1. 检查是否已抢购（防重）
if redis.call("EXISTS", "seckill:user:" .. userId .. ":product:" .. string.match(key, ":(%d+)$")) == 1 then
    return -2  -- 已参与
end
-- 2. 获取当前库存
local stock = tonumber(redis.call("GET", key))
if not stock or stock <= 0 then
    return -1  -- 库存不足
end
-- 3. 扣减库存
redis.call("DECR", key)
-- 4. 记录用户已参与（防重，TTL 可选）
redis.call("SET", "seckill:user:" .. userId .. ":product:" .. string.match(key, ":(%d+)$"), "1", "EX", 3600)
return stock - 1

// main.go 或 handler/seckill.go
func SeckillHandler(c *gin.Context) {
    userID := c.GetString("user_id") // 假设已鉴权
    productID := c.Param("product_id")
    // 构造 Redis Key
    stockKey := fmt.Sprintf("stock:product:%s", productID)
    userProductKey := fmt.Sprintf("seckill:user:%s:product:%s", userID, productID)
    // 执行 Lua 脚本
    result, err := redisClient.Eval(
        ctx,
        luaScript,           // 上述 Lua 脚本内容
        []string{stockKey},
        userID,
    ).Result()
    if err != nil {
        c.JSON(500, gin.H{"error": "系统繁忙"})
        return
    }
    switch ret := result.(type) {
    case int64:
        if ret == -1 {
            c.JSON(400, gin.H{"error": "库存不足"})
            return
        }
        if ret == -2 {
            c.JSON(400, gin.H{"error": "您已参与过本次秒杀"})
            return
        }
    default:
        c.JSON(500, gin.H{"error": "未知错误"})
        return
    }
    // 成功！发送消息到 MQ（异步落库）
    msg := SeckillMessage{
        UserID:    userID,
        ProductID: productID,
        Timestamp: time.Now(),
    }
    // 序列化并发送到 Kafka / RabbitMQ
    if err := mqProducer.Send("seckill_queue", msg); err != nil {
        // 注意：此处即使 MQ 发送失败，Redis 已扣减，需有补偿机制！
        log.Printf("MQ send failed: %v", err)
        // 可考虑回滚 Redis（复杂），或依赖后续对账
    }
    // 立即返回用户“抢购成功，请等待订单生成”
    c.JSON(200, gin.H{
        "msg": "抢购成功！正在生成订单...",
        "queue_status": "processing",
    })
}

// worker/seckill_worker.go
func StartSeckillWorker() {
    for msg := range mqConsumer.Subscribe("seckill_queue") {
        var seckillMsg SeckillMessage
        if err := json.Unmarshal(msg, &seckillMsg); err != nil {
            continue
        }
        // 开启事务，落库
        tx := db.Begin()
        defer tx.Rollback()
        // 1. 再次校验（兜底）：MySQL 中库存是否足够？
        var product Product
        if err := tx.Where("id = ? AND stock > 0", seckillMsg.ProductID).First(&product).Error; err != nil {
            log.Printf("MySQL 库存不足或商品不存在: %v", seckillMsg)
            continue // 丢弃消息 or DLQ
        }
        // 2. 创建订单
        order := Order{
            UserID:    seckillMsg.UserID,
            ProductID: seckillMsg.ProductID,
            Status:    "created",
        }
        if err := tx.Create(&order).Error != nil {
            continue
        }
        // 3. 扣减 MySQL 库存
        if err := tx.Model(&Product{}).
            Where("id = ? AND stock = ?", seckillMsg.ProductID, product.Stock).
            Update("stock", gorm.Expr("stock - 1")).Error; err != nil {
            continue
        }
        tx.Commit()
        log.Printf("订单创建成功: %v", order.ID)
    }
}