Redis 分布式锁

一、定义

1. 基本概念

Redis 分布式锁是基于 Redis 单线程特性实现的分布式协同工具，用于解决多进程 / 多节点间的资源竞争问题。

二、特点

1. 轻量高效

• 基于内存操作，性能远高于数据库锁；

2. 原子性

• 通过 Redis 原子命令 / Lua 脚本保证加锁、解锁操作的原子性；

3. 可扩展

• 支持单节点、多节点（Redlock）部署，适配不同可用性要求；

4. 易集成

• 支持多种客户端（Go/Java/Python），适配分布式系统架构。

三、核心特性

1. 互斥性

• 核心特性，同一时刻仅允许一个客户端持有锁，避免资源竞争

2. 原子性

• 加锁、解锁操作必须原子化，避免中间态导致的锁异常

3. 超时释放

• 避免客户端持有锁后宕机导致死锁，需设置合理的锁超时时间

4. 高可用性

• 避免 Redis 单点故障导致锁服务不可用，支持多节点部署（Redlock）

5. 可重入性

• 可选特性，同一客户端多次获取同一把锁不阻塞（需额外实现）

6. 公平性

• 可选特性，按请求顺序获取锁（Redis 原生不支持，需额外设计）

7. 容错性

• 部分 Redis 节点故障时，锁服务仍能正常工作（Redlock 保证）

四、实现原理

1. 核心基础

Redis 分布式锁的核心依赖其 单线程模型（命令串行执行）和 原子命令，核心逻辑：

• 加锁：通过 SET key value NX EX seconds 命令实现原子加锁（NX = 仅当 key 不存在时设置，EX = 自动过期时间）；
• 解锁：通过 Lua 脚本实现「判断唯一标识 + 删除锁」的原子操作，避免误删其他客户端的锁；
• 高可用：Redlock 算法通过多 Redis 实例加锁（超过半数实例加锁成功则认为锁有效），解决单点故障问题

2. 核心命令解析

• SET lock_key unique_val NX EX 10：原子加锁，lock_key 为锁名，unique_val 为客户端唯一标识，NX 保证互斥，EX 设置 10s 超时
• Lua 脚本解锁：原子判断 unique_val 是否匹配，匹配则删除锁，避免误删

五、业务场景

1. 电商超卖 / 库存扣减

• 场景：多节点并发扣减库存时，避免库存数量为负；
• 核心：通过分布式锁保证库存扣减操作的原子性。

2. 秒杀活动

• 场景：限制商品秒杀的并发抢购，避免超量下单；
• 核心：锁粒度控制（商品维度），避免全量锁导致性能瓶颈。

3. 分布式任务调度

• 场景：多节点定时任务避免重复执行（如数据同步、报表生成）；
• 核心：通过锁保证同一任务仅一个节点执行

4. 订单状态更新

• 场景：多系统并发更新订单状态（如支付、发货），避免状态不一致；
• 核心：锁绑定订单 ID，保证状态更新的互斥性。

5. 缓存更新

• 场景：避免多节点同时更新缓存导致的缓存击穿；
• 核心：加锁后更新缓存，未获取锁的节点等待或直接读取数据库

六、电商超卖场景：分布式锁渐进式实现（Go）

1. 场景背景

电商商品库存为 100，多节点并发下单扣减库存，若未做并发控制，会出现库存为负（超卖）。以下通过 6 个版本逐步优化分布式锁实现。

2. 依赖准备

package main

import (
	"context"
	"crypto/rand"
	"encoding/base64"
	"errors"
	"fmt"
	"github.com/go-redis/redis/v8"
	"math/big"
	"sync"
	"time"
)

// 全局变量
var (
	// Redis 客户端（Redis 8.4 兼容 go-redis/v8）
	redisClient = redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "",
		DB:       0,
	})
	ctx = context.Background()
	// 模拟库存（实际应存储在 Redis/数据库）
	stock = 100
	stockMu sync.Mutex // 本地锁保护库存变量
)

// 生成客户端唯一标识（避免误删锁）
func generateUniqueID() string {
	b := make([]byte, 16)
	rand.Read(b)
	return base64.URLEncoding.EncodeToString(b)
}

3. 版本 1：基础 SETNX 实现（存在死锁风险）

（1）代码实现

// 加锁（版本1：仅SETNX，无超时）
func lockV1(lockKey string) bool {
	// SETNX：key不存在时返回1，存在返回0
	result, err := redisClient.SetNX(ctx, lockKey, "default", 0).Result()
	if err != nil {
		fmt.Println("加锁失败：", err)
		return false
	}
	return result
}

// 解锁（版本1：直接DEL）
func unlockV1(lockKey string) {
	_, err := redisClient.Del(ctx, lockKey).Result()
	if err != nil {
		fmt.Println("解锁失败：", err)
	}
}

// 扣减库存（版本1）
func deductStockV1() {
	lockKey := "lock:stock:1001"
	// 加锁
	if !lockV1(lockKey) {
		fmt.Println("获取锁失败，放弃扣减")
		return
	}

	// 业务逻辑：扣减库存
	stockMu.Lock()
	if stock > 0 {
		stock--
		fmt.Printf("库存扣减成功，剩余库存：%d\n", stock)
	} else {
		fmt.Println("库存不足，扣减失败")
	}
	stockMu.Unlock()

	// 解锁
	unlockV1(lockKey)
}

（2）问题分析

• 死锁风险：若加锁后服务宕机 / 进程崩溃，DEL 未执行，锁永远无法释放，导致后续所有请求无法获取锁；
• 无超时机制：锁没有自动过期，一旦异常则死锁
• 加锁时，缺少自旋机制（多次加锁重试）

4. 版本 2：SETNX + EXPIRE（非原子，仍有问题）

（1）代码实现

// 加锁（版本2：SETNX + EXPIRE）
func lockV2(lockKey string, expire int) bool {
	// 第一步：SETNX加锁
	result, err := redisClient.SetNX(ctx, lockKey, "default", 0).Result()
	if err != nil || !result {
		fmt.Println("加锁失败：", err)
		return false
	}

	// 第二步：设置过期时间
	_, err = redisClient.Expire(ctx, lockKey, time.Duration(expire)*time.Second).Result()
	if err != nil {
		fmt.Println("设置过期时间失败：", err)
		// 回滚：删除已加的锁
		redisClient.Del(ctx, lockKey)
		return false
	}
	return true
}

// 扣减库存（版本2）
func deductStockV2() {
	lockKey := "lock:stock:1001"
	// 加锁（超时10s）
	if !lockV2(lockKey, 10) {
		fmt.Println("获取锁失败，放弃扣减")
		return
	}

	// 业务逻辑：扣减库存
	stockMu.Lock()
	if stock > 0 {
		stock--
		fmt.Printf("库存扣减成功，剩余库存：%d\n", stock)
	} else {
		fmt.Println("库存不足，扣减失败")
	}
	stockMu.Unlock()

	// 解锁
	unlockV1(lockKey)
}

（2）优化点

• 增加超时过期

（3）问题分析

• 非原子操作：SETNX 和 EXPIRE 是两个独立命令，若 SETNX 成功后、EXPIRE 执行前服务宕机，仍会导致锁无超时，引发死锁；
• 无唯一标识：解锁时直接 DEL，可能误删其他客户端持有的锁（比如锁超时自动释放后，其他客户端加锁，当前客户端执行 DEL）

5. 版本 3：原子 SET NX EX + 唯一标识（解决原子加锁 + 避免误删）

（1）代码实现

// 加锁（版本3：原子SET NX EX + 唯一标识）
func lockV3(lockKey string, expire int) (string, bool) {
	// 生成客户端唯一标识
	uniqueID := generateUniqueID()
	// SET key value NX EX seconds：原子加锁+超时+唯一标识
	result, err := redisClient.Set(ctx, lockKey, uniqueID, time.Duration(expire)*time.Second).SetNX().Result()
	if err != nil || !result {
		fmt.Println("加锁失败：", err)
		return "", false
	}
	return uniqueID, true
}

// 解锁（版本3：先判断唯一标识，再删除）
func unlockV3(lockKey, uniqueID string) {
	// 第一步：获取锁的value
	val, err := redisClient.Get(ctx, lockKey).Result()
	if err != nil {
		fmt.Println("获取锁标识失败：", err)
		return
	}

	// 第二步：判断是否为当前客户端的标识
	if val == uniqueID {
		// 第三步：删除锁
		_, err = redisClient.Del(ctx, lockKey).Result()
		if err != nil {
			fmt.Println("解锁失败：", err)
		}
	} else {
		fmt.Println("锁标识不匹配，拒绝解锁")
	}
}

// 扣减库存（版本3）
func deductStockV3() {
	lockKey := "lock:stock:1001"
	// 加锁（超时10s）
	uniqueID, ok := lockV3(lockKey, 10)
	if !ok {
		fmt.Println("获取锁失败，放弃扣减")
		return
	}

	// 业务逻辑：扣减库存
	stockMu.Lock()
	if stock > 0 {
		stock--
		fmt.Printf("库存扣减成功，剩余库存：%d\n", stock)
	} else {
		fmt.Println("库存不足，扣减失败")
	}
	stockMu.Unlock()

	// 解锁
	unlockV3(lockKey, uniqueID)
}

（2）优化点

• 用 SET NX EX 原子命令替代 SETNX + EXPIRE，解决加锁 + 超时的原子性；
• 引入唯一标识，解锁前先校验，避免误删其他客户端的锁。

（3）问题分析

• 解锁操作（GET + DEL）非原子：若 GET 后、DEL 前，锁超时自动释放，其他客户端加锁，当前客户端仍会执行 DEL，误删新锁；
• Redis 单点故障：若 Redis 节点宕机，锁服务不可用。

6. 版本 4：Lua 脚本原子解锁（解决解锁原子性）

（1）代码实现

// 解锁（版本4：Lua脚本原子解锁）
func unlockV4(lockKey, uniqueID string) error {
	// Lua脚本：判断value匹配则删除，保证原子性
	unlockScript := `
		if redis.call('get', KEYS[1]) == ARGV[1] then
			return redis.call('del', KEYS[1])
		else
			return 0
		end
	`
	// 执行Lua脚本
	result, err := redisClient.Eval(ctx, unlockScript, []string{lockKey}, uniqueID).Result()
	if err != nil {
		return fmt.Errorf("执行解锁脚本失败：%w", err)
	}
	if res, ok := result.(int64); !ok || res == 0 {
		return errors.New("锁标识不匹配或锁已过期，解锁失败")
	}
	return nil
}

// 扣减库存（版本4）
func deductStockV4() {
	lockKey := "lock:stock:1001"
	// 加锁（超时10s）
	uniqueID, ok := lockV3(lockKey, 10)
	if !ok {
		fmt.Println("获取锁失败，放弃扣减")
		return
	}

	// 模拟业务耗时（比如网络请求）
	time.Sleep(2 * time.Second)

	// 业务逻辑：扣减库存
	stockMu.Lock()
	if stock > 0 {
		stock--
		fmt.Printf("库存扣减成功，剩余库存：%d\n", stock)
	} else {
		fmt.Println("库存不足，扣减失败")
	}
	stockMu.Unlock()

	// 解锁
	if err := unlockV4(lockKey, uniqueID); err != nil {
		fmt.Println("解锁失败：", err)
	}
}

（2）优化点

• 用 Lua 脚本将「判断标识 + 删除锁」封装为原子操作，彻底解决误删锁问题

（3）问题分析

• 锁超时问题：若业务执行时间超过锁超时时间（比如耗时 15s，锁超时 10s），锁会自动释放，其他客户端加锁，导致并发扣减；
• Redis 单点故障：Redis 主节点宕机，从节点未同步锁数据，导致锁失效

7. 版本 5：Redlock 算法（解决 Redis 单点故障）

（1）代码实现

// Redlock 分布式锁结构体
type RedLock struct {
	clients []*redis.Client // 多Redis实例客户端
	quorum  int             // 最小成功节点数（len(clients)/2 + 1）
}

// NewRedLock 初始化Redlock
func NewRedLock(addrs []string) *RedLock {
	clients := make([]*redis.Client, len(addrs))
	for i, addr := range addrs {
		clients[i] = redis.NewClient(&redis.Options{
			Addr:     addr,
			Password: "",
			DB:       0,
		})
	}
	return &RedLock{
		clients: clients,
		quorum:  len(clients)/2 + 1,
	}
}

// Lock Redlock加锁
func (rl *RedLock) Lock(lockKey string, expire int) (string, bool) {
	uniqueID := generateUniqueID()
	successCount := 0
	// 记录开始时间（用于判断超时）
	start := time.Now()

	// 遍历所有Redis实例加锁
	for _, client := range rl.clients {
		result, err := client.Set(ctx, lockKey, uniqueID, time.Duration(expire)*time.Second).SetNX().Result()
		if err == nil && result {
			successCount++
		}
		// 超过半数实例加锁成功，且总耗时 < 锁超时时间的1/3，认为加锁成功
		if successCount >= rl.quorum && time.Since(start) < time.Duration(expire)*time.Second/3 {
			return uniqueID, true
		}
	}

	// 加锁失败，回滚已加锁的节点
	if successCount > 0 {
		rl.Unlock(lockKey, uniqueID)
	}
	return "", false
}

// Unlock Redlock解锁
func (rl *RedLock) Unlock(lockKey, uniqueID string) {
	unlockScript := `
		if redis.call('get', KEYS[1]) == ARGV[1] then
			return redis.call('del', KEYS[1])
		else
			return 0
		end
	`
	// 遍历所有Redis实例解锁
	for _, client := range rl.clients {
		client.Eval(ctx, unlockScript, []string{lockKey}, uniqueID)
	}
}

// 扣减库存（版本5：Redlock）
func deductStockV5(rl *RedLock) {
	lockKey := "lock:stock:1001"
	// 加锁（超时10s）
	uniqueID, ok := rl.Lock(lockKey, 10)
	if !ok {
		fmt.Println("获取Redlock失败，放弃扣减")
		return
	}

	// 模拟业务耗时
	time.Sleep(2 * time.Second)

	// 业务逻辑：扣减库存
	stockMu.Lock()
	if stock > 0 {
		stock--
		fmt.Printf("库存扣减成功，剩余库存：%d\n", stock)
	} else {
		fmt.Println("库存不足，扣减失败")
	}
	stockMu.Unlock()

	// 解锁
	rl.Unlock(lockKey, uniqueID)
}

（2）优化点

• 基于 Redlock 算法，多 Redis 实例（至少 3 个独立节点）加锁，超过半数节点加锁成功则认为锁有效，解决单点故障问题。

（3）问题分析

• 锁超时问题仍存在；
• Redlock 部署成本高（需多个独立 Redis 节点），性能略低于单节点。

8. 版本 6：带看门狗的可重入锁（解决锁超时 + 支持重入）

（1）代码实现

// 可重入分布式锁结构体（带看门狗）
type ReentrantLock struct {
	client     *redis.Client
	lockKey    string
	uniqueID   string        // 客户端唯一标识
	expire     time.Duration // 锁基础超时
	count      int           // 重入次数
	watchdog   *time.Ticker  // 看门狗定时器
	ctx        context.Context
	cancelFunc context.CancelFunc
}

// NewReentrantLock 初始化可重入锁
func NewReentrantLock(client *redis.Client, lockKey string, expire time.Duration) *ReentrantLock {
	ctx, cancel := context.WithCancel(context.Background())
	return &ReentrantLock{
		client:     client,
		lockKey:    lockKey,
		uniqueID:   generateUniqueID(),
		expire:     expire,
		ctx:        ctx,
		cancelFunc: cancel,
	}
}

// Lock 加锁（支持重入）
func (rl *ReentrantLock) Lock() bool {
	// 重入判断：若已持有锁，直接增加重入次数
	if rl.count > 0 {
		rl.count++
		return true
	}

	// 首次加锁：原子SET NX EX
	result, err := rl.client.Set(ctx, rl.lockKey, rl.uniqueID, rl.expire).SetNX().Result()
	if err != nil || !result {
		return false
	}

	// 启动看门狗：每 expire/3 时间续期一次
	rl.count = 1
	rl.watchdog = time.NewTicker(rl.expire / 3)
	go func() {
		for {
			select {
			case <-rl.watchdog.C:
				// Lua脚本：判断标识匹配则续期
				renewScript := `
					if redis.call('get', KEYS[1]) == ARGV[1] then
						return redis.call('expire', KEYS[1], ARGV[2])
					else
						return 0
					end
				`
				rl.client.Eval(ctx, renewScript, []string{rl.lockKey}, rl.uniqueID, int(rl.expire.Seconds()))
			case <-rl.ctx.Done():
				rl.watchdog.Stop()
				return
			}
		}
	}()
	return true
}

// Unlock 解锁（支持重入）
func (rl *ReentrantLock) Unlock() error {
	// 重入次数减1
	if rl.count > 1 {
		rl.count--
		return nil
	}

	// 最后一次解锁：停止看门狗 + 原子删除锁
	rl.cancelFunc()
	unlockScript := `
		if redis.call('get', KEYS[1]) == ARGV[1] then
			return redis.call('del', KEYS[1])
		else
			return 0
		end
	`
	result, err := rl.client.Eval(ctx, unlockScript, []string{rl.lockKey}, rl.uniqueID).Result()
	if err != nil {
		return err
	}
	if res, ok := result.(int64); !ok || res == 0 {
		return errors.New("锁已过期或标识不匹配")
	}
	rl.count = 0
	return nil
}

// 扣减库存（版本6：可重入+看门狗）
func deductStockV6(rl *ReentrantLock) {
	// 加锁
	if !rl.Lock() {
		fmt.Println("获取可重入锁失败，放弃扣减")
		return
	}

	// 模拟重入（比如嵌套调用）
	rl.Lock()

	// 模拟长耗时业务（超过基础超时，看门狗自动续期）
	time.Sleep(15 * time.Second)

	// 业务逻辑：扣减库存
	stockMu.Lock()
	if stock > 0 {
		stock--
		fmt.Printf("库存扣减成功，剩余库存：%d\n", stock)
	} else {
		fmt.Println("库存不足，扣减失败")
	}
	stockMu.Unlock()

	// 解锁（重入次数2，需解锁2次）
	if err := rl.Unlock(); err != nil {
		fmt.Println("第一次解锁失败：", err)
	}
	if err := rl.Unlock(); err != nil {
		fmt.Println("第二次解锁失败：", err)
	}
}

（2）优化点

• 看门狗机制：定时（锁超时的 1/3）自动续期锁，解决业务耗时超过锁超时的问题；
• 可重入性：通过本地计数 + 唯一标识，支持同一客户端多次加锁；
  • 也可以用 hash 解决可重入性
• 原子操作：加锁 / 解锁 / 续期均通过原子命令 / Lua 脚本实现。

9. 测试代码（模拟并发超卖）

func main() {
	// 模拟100个并发请求扣减库存
	var wg sync.WaitGroup
	wg.Add(100)

	// 版本6测试（可重入+看门狗）
	rl := NewReentrantLock(redisClient, "lock:stock:1001", 10*time.Second)
	for i := 0; i < 100; i++ {
		go func() {
			defer wg.Done()
			deductStockV6(rl)
		}()
	}

	wg.Wait()
	fmt.Printf("最终库存：%d\n", stock) // 最终库存应为0，无超卖
}

七、Redlock

1. 现状

• Redis 官方态度：Redis 作者 Salvatore Sanfilippo（antirez）明确表示 Redlock 存在设计缺陷，不推荐在高一致性要求的场景使用
• 社区实践：主流分布式锁框架（如 Redisson）仍保留 Redlock 实现，但标注为「非强一致性」方案，且更多场景优先推荐单节点 + 主从 / 哨兵架构
• 适用场景收缩：Redlock 仅在「Redis 单节点不可用、且无法接受主从切换导致的锁短暂失效」的小众场景中少量使用，绝大多数业务场景无需依赖

2. 争议的核心原因

核心缺陷使其无法保证「严格的分布式一致性」。

• 时钟漂移问题：Redlock 依赖多个 Redis 节点的「时间一致性」；若某节点时钟跳变（如手动调整、NTP 同步异常），可能导致锁超时时间失效，进而引发多客户端同时持有锁，破坏互斥性
• 性能与复杂度问题：
  • Redlock 需向至少 3 个独立 Redis 节点发起加锁请求，网络耗时是单节点锁的 3 倍以上，高并发场景下性能损耗显著；
  • 部署成本高（需独立 Redis 节点，不能复用主从集群），运维复杂度提升。
• 一致性模型缺陷：Redlock 试图用「过半节点加锁成功」模拟分布式一致性，但未遵循严格的分布式一致性协议（如 Paxos/Raft），在节点网络分区、超时等异常场景下，仍可能出现锁失效

3. 替代案例：多重锁

八、常见问题

1. 死锁

• 原因：加锁后服务宕机，锁无超时 / 超时未生效；
• 解决方案：使用 SET NX EX 原子命令设置超时，避免手动 SETNX + EXPIRE

2. 误删锁

• 原因：解锁时未校验唯一标识，或 GET + DEL 非原子；
• 解决方案：加锁时设置唯一标识，解锁时用 Lua 脚本原子校验 + 删除

3. 锁超时

• 原因：业务执行时间超过锁超时，锁自动释放；
• 解决方案：看门狗定时续期，或预估合理的超时时间（略大于业务最大耗时）

4. 单点故障

• 原因：Redis 单节点宕机，锁服务不可用；
• 解决方案：Redlock 多节点部署，或 Redis 主从 + 哨兵架构

5. 重入问题

• 原因：同一客户端多次加锁导致死锁；
• 解决方案：实现可重入锁（本地计数 + 唯一标识）

6. 公平性问题

• 原因：Redis 分布式锁为非公平锁，可能导致某些客户端长期获取不到锁；
• 解决方案：结合有序集合（ZSET）实现公平锁（按请求时间排队）
  • score 记录请求时间戳

九、注意事项

1. 锁粒度设计

• 避免「全量锁」（如整个库存系统一把锁），应按资源维度拆分（如商品 ID 维度），降低锁竞争；
• 示例：锁名设计为 lock:stock:{商品ID}，而非 lock:stock。

2. 超时时间设置

• 超时时间需略大于业务最大执行时间（比如业务耗时 5s，超时设为 10s）；
• 避免超时过短（导致锁提前释放）或过长（死锁时影响范围大）

3. 看门狗实现

• 看门狗需在单独 goroutine 中运行，避免阻塞业务；
• 解锁时需停止看门狗，避免无效续期

4. Redlock 适用场景

• Redlock 适合对可用性要求极高的场景（如金融、电商核心交易）；
• 普通场景可使用 Redis 主从 + 哨兵，无需过度设计

5. 异常处理

• 加锁失败时，应重试（带退避策略）而非直接放弃；
• 解锁失败时，需记录日志并监控，避免锁残留

十、面试题

1. 分布式锁的核心特性有哪些？Redis 如何保证这些特性？

（1）核心特性

• 互斥性、原子性、超时释放、高可用、可重入

（2）Redis 实现

• 互斥性：SET NX 命令保证同一时刻仅一个客户端加锁；
• 原子性：SET NX EX 原子命令、Lua 脚本保证加锁 / 解锁原子性；
• 超时释放：EX 参数设置自动过期，避免死锁；
• 高可用：Redlock 多节点部署，或主从 + 哨兵架构；
• 可重入：本地计数 + 唯一标识，多次加锁仅增加计数

2. Redis 分布式锁和 Zookeeper 分布式锁的对比？

（1）对比

3. Redis 实现分布式锁时，为什么要用 Lua 脚本解锁？直接 GET + DEL 不行吗？

• 直接 GET + DEL 非原子：若 GET 后、DEL 前，锁超时自动释放，其他客户端加锁，当前客户端会误删新锁；
• Lua 脚本将「判断标识 + 删除锁」封装为原子操作，Redis 单线程执行脚本，避免中间态，彻底解决误删问题

4. Redlock 算法的核心原理？有哪些缺陷？

（1）核心原理

• 部署至少 3 个独立 Redis 节点；
• 客户端向所有节点发送加锁请求；
• 超过半数节点加锁成功，且总耗时 < 锁超时时间的 1/3，则认为加锁成功；
• 解锁时向所有节点发送解锁请求

（2）缺陷

• 部署成本高（需多个独立节点）；
• 性能低于单节点；
• 时钟漂移可能导致锁失效（节点时间不一致）

5. 如何解决 Redis 分布式锁的超时问题？

• 看门狗机制：客户端启动定时任务，每锁超时的 1/3 时间续期一次（Lua 脚本原子续期）；
• 预估合理超时：基于业务压测结果，设置略大于最大执行时间的超时；
• 分段锁：将长耗时业务拆分为多个短任务，每个任务加锁执行，降低单次锁持有时间

6. 电商超卖问题的根本原因是什么？如何用分布式锁彻底解决？

（1）根本原因

• 多节点并发扣减库存时，未保证库存操作的原子性（读 - 改 - 写非原子）

（2）分布式锁解决方案

• 锁粒度：按商品 ID 加锁，避免全量锁；
• 原子加锁：SET NX EX 原子命令，避免死锁；
• 原子解锁：Lua 脚本校验唯一标识后删除，避免误删；
• 看门狗续期：解决业务耗时超过锁超时的问题；
• 兜底校验：扣减库存前再次校验库存是否大于 0，避免极端情况超卖

7. Redis 分布式锁在高并发场景下的性能优化手段？

• 锁粒度拆分：按资源维度拆分锁（如商品 ID、用户 ID），降低锁竞争；
• 本地缓存锁：热点资源的锁可本地缓存，减少 Redis 访问；
• 批量解锁 / 加锁：合并多个锁操作，减少 Redis 网络 IO；
• 非阻塞加锁：加锁失败时直接返回，而非阻塞等待，避免线程阻塞；
• 连接池优化：增大 Redis 连接池，避免连接耗尽导致的性能瓶颈

8. Redis 分布式锁的可重入性如何实现？有哪些注意事项？

（1）实现方式

• 加锁时生成唯一标识（客户端 ID + 线程 ID）；
• 本地维护重入计数，首次加锁计数为 1，重复加锁计数 + 1；
• 解锁时计数 - 1，计数为 0 时才执行 Redis DEL 操作

（2）注意事项

• 唯一标识需全局唯一，避免不同客户端的重入计数冲突；
• 看门狗续期需与重入计数绑定，避免续期已释放的锁；
• 进程崩溃时，本地计数丢失，需依赖 Redis 超时释放锁。

9. 如果 Redis 分布式锁失效，有哪些兜底方案？

• 数据库乐观锁：库存表增加版本号，扣减时 WHERE 库存>0 AND 版本号=xxx，保证原子性；
• Redis 原子操作：用 DECRBY 原子扣减库存，扣减前判断结果是否 >=0；
• 限流兜底：网关层限制并发请求数，避免超出系统处理能力；
• 监控告警：实时监控库存数量，超卖时立即告警并熔断下单接口。