# 分布式锁

## 问题

1. 什么是分布式锁？为什么需要分布式锁？
2. 如何用 Redis 实现分布式锁？
3. Redis 分布式锁有哪些坑？如何解决？
4. Zookeeper 如何实现分布式锁？
5. Redis 分布式锁和 Zookeeper 分布式锁的区别？
6. Redisson 是如何实现分布式锁的？
7. 在实际项目中如何使用分布式锁？

---

## 标准答案

### 1. 为什么需要分布式锁

#### **场景**

**单机锁（synchronized、ReentrantLock）**：
```java
@Service
public class OrderService {
    private synchronized void createOrder(Order order) {
        // 单机环境下有效
        // 分布式环境下无效（不同 JVM）
    }
}
```

**问题**：
- 单机锁只对单个 JVM 有效
- 分布式环境下，多个实例的锁互不排斥
- 需要跨 JVM、跨机器的锁机制

---

### 2. Redis 分布式锁

#### **基本实现**

```java
public class RedisDistributedLock {

    private final StringRedisTemplate redisTemplate;

    // 加锁
    public boolean lock(String key, String value, long expireTime) {
        // SET key value NX PX expireTime
        Boolean success = redisTemplate.opsForValue()
            .setIfAbsent(key, value, expireTime, TimeUnit.MILLISECONDS);

        return success != null && success;
    }

    // 释放锁
    public boolean unlock(String key, String value) {
        // Lua 脚本：保证原子性
        String luaScript =
            "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "  return redis.call('del', KEYS[1]) " +
            "else " +
            "  return 0 " +
            "end";

        DefaultRedisScript<Long> script = new DefaultRedisScript<>(luaScript, Long.class);
        Long result = redisTemplate.execute(script, Collections.singletonList(key), value);

        return result != null && result == 1;
    }
}
```

---

#### **使用示例**

```java
@Service
public class OrderService {

    @Autowired
    private RedisDistributedLock lock;

    public void createOrder(Order order) {
        String lockKey = "order:" + order.getProductId();
        String lockValue = UUID.randomUUID().toString();

        try {
            // 加锁（过期时间 30 秒）
            boolean locked = lock.lock(lockKey, lockValue, 30000);

            if (!locked) {
                throw new BusinessException("系统繁忙，请稍后再试");
            }

            // 执行业务逻辑
            // 1. 查询库存
            // 2. 扣减库存
            // 3. 创建订单

        } finally {
            // 释放锁
            lock.unlock(lockKey, lockValue);
        }
    }
}
```

---

### 3. Redis 分布式锁的坑

#### **坑 1：锁超时问题**

**场景**：
```
线程 A 获取锁（设置 30 秒过期）
线程 A 执行业务（耗时 50 秒）
30 秒后，锁自动过期
线程 B 获取锁
线程 A 执行完毕，释放锁（把线程 B 的锁释放了！）
```

**解决方案：看门狗（Watchdog）**

**原理**：
- 启动后台线程
- 定期检查锁是否还存在
- 如果存在，续期（延长过期时间）

**Redisson 实现**：
```java
RLock lock = redisson.getLock("myLock");
lock.lock();  // 自动续期（默认 30 秒）

try {
    // 业务逻辑
} finally {
    lock.unlock();
}
```

---

#### **坑 2：主从切换导致锁丢失**

**场景**：
```
线程 A 在 Master 获取锁
锁未同步到 Slave
Master 宕机，Slave 升级为 Master
线程 B 在新 Master 上获取锁（冲突！）
```

**解决方案：Redlock（Redis 分布式锁算法）**

**原理**：
- 向多个独立的 Redis 节点请求加锁
- 大多数节点加锁成功才算成功

```java
// Redisson Redlock
RLock lock1 = redisson1.getLock("myLock");
RLock lock2 = redisson2.getLock("myLock");
RLock lock3 = redisson3.getLock("myLock");

RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);

try {
    redLock.lock();
    // 业务逻辑
} finally {
    redLock.unlock();
}
```

---

#### **坑 3：释放锁时误删**

**场景**：
```
线程 A 获取锁
线程 A 执行时间过长，锁超时过期
线程 B 获取锁
线程 A 执行完毕，释放锁（删除了线程 B 的锁！）
```

**解决方案：Lua 脚本 + 唯一标识**

```lua
-- Lua 脚本（原子操作）
if redis.call('get', KEYS[1]) == ARGV[1] then
  return redis.call('del', KEYS[1])
else
  return 0
end
```

**Java 实现**：
```java
public boolean unlock(String key, String value) {
    String luaScript =
        "if redis.call('get', KEYS[1]) == ARGV[1] then " +
        "  return redis.call('del', KEYS[1]) " +
        "else " +
        "  return 0 " +
        "end";

    DefaultRedisScript<Long> script = new DefaultRedisScript<>(luaScript, Long.class);
    Long result = redisTemplate.execute(script, Collections.singletonList(key), value);

    return result != null && result == 1;
}
```

---

### 4. Zookeeper 分布式锁

#### **原理**

利用 Zookeeper 的**临时顺序节点**：

```
/lock
  /lock/node-0000000001  (客户端 1)
  /lock/node-0000000002  (客户端 2)
  /lock/node-0000000003  (客户端 3)
```

**流程**：
1. 客户端创建临时顺序节点
2. 获取所有子节点，判断自己是否是序号最小的
3. 如果是最小节点，获取锁
4. 如果不是，监听前一个节点的删除事件

---

#### **Curator 实现**

```java
@Component
public class ZkDistributedLock {

    private final InterProcessMutex lock;

    public ZkDistributedLock(CuratorFramework curatorFramework) {
        this.lock = new InterProcessMutex(curatorFramework, "/lock");
    }

    public void acquire() {
        try {
            lock.acquire();
        } catch (Exception e) {
            throw new RuntimeException("获取锁失败", e);
        }
    }

    public void release() {
        try {
            lock.release();
        } catch (Exception e) {
            throw new RuntimeException("释放锁失败", e);
        }
    }
}
```

**使用**：
```java
@Service
public class OrderService {

    @Autowired
    private ZkDistributedLock lock;

    public void createOrder(Order order) {
        try {
            lock.acquire();

            // 业务逻辑

        } finally {
            lock.release();
        }
    }
}
```

---

### 5. Redis vs Zookeeper

| 特性        | Redis        | Zookeeper        |     |
| --------- | ------------ | ---------------- | --- |
| **性能**    | 高（内存操作）      | 低（磁盘 + ZAB 协议）   |     |
| **可靠性**   | 中（可能丢锁）      | 高（CP 一致性）        |     |
| **实现复杂度** | 简单           | 复杂               |     |
| **获取锁方式** | 轮询           | Watcher 通知（事件驱动） |     |
| **锁释放**   | 超时自动释放       | 会话结束自动释放         |     |
| **适用场景**  | 高并发、对一致性要求不高 | 严格一致性要求          |     |

---

### 6. Redisson 原理

#### **看门狗（自动续期）**

```java
RLock lock = redisson.getLock("myLock");
lock.lock();  // 默认 leaseTime = -1（启用看门狗）
```

**原理**：
```
1. 加锁成功，启动看门狗线程
2. 看门狗每 10 秒检查一次
3. 如果锁还存在，续期 30 秒
4. 客户端宕机，会话结束，看门狗停止，锁自动过期
```

---

#### **公平锁**

```java
RLock fairLock = redisson.getFairLock("myLock");
fairLock.lock();
```

**原理**：
- 加锁时，添加到队列尾部
- 按照请求顺序获取锁
- 性能比非公平锁低

---

#### **读写锁**

```java
RReadWriteLock rwLock = redisson.getReadWriteLock("myLock");
rwLock.readLock().lock();   // 读锁（共享）
rwLock.writeLock().lock();  // 写锁（独占）
```

---

### 7. 实际项目应用

#### **场景 1：秒杀系统**

```java
@Service
public class SeckillService {

    @Autowired
    private RedissonClient redisson;

    public void seckill(Long userId, Long productId) {
        RLock lock = redisson.getLock("seckill:" + productId);

        try {
            // 尝试加锁（最多等待 3 秒，锁自动释放时间 10 秒）
            boolean locked = lock.tryLock(3, 10, TimeUnit.SECONDS);

            if (!locked) {
                throw new BusinessException("抢购人数过多，请稍后再试");
            }

            // 1. 查询库存
            Integer stock = redisTemplate.opsForValue().get("stock:" + productId);
            if (stock == null || stock <= 0) {
                throw new BusinessException("库存不足");
            }

            // 2. 扣减库存
            redisTemplate.opsForValue().decrement("stock:" + productId);

            // 3. 创建订单
            createOrder(userId, productId);

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("系统异常");
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}
```

---

#### **场景 2：定时任务分布式锁**

```java
@Component
public class ScheduledTask {

    @Autowired
    private RedissonClient redisson;

    @Scheduled(cron = "0 */5 * * * ?")  // 每 5 分钟
    public void execute() {
        RLock lock = redisson.getLock("scheduled-task");

        try {
            // 只有一个实例执行
            boolean locked = lock.tryLock(0, 5, TimeUnit.MINUTES);

            if (!locked) {
                return;  // 其他实例正在执行
            }

            // 执行任务
            doTask();

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}
```

---

### 8. 阿里 P7 加分项

**深度理解**：
- 理解 CAP 理论在分布式锁中的体现
- 理解 Redis 的 ZSet 实现延迟队列的原理
- 理解 ZAB 协议和 ZK 的一致性保证

**实战经验**：
- 有处理分布式锁死锁的经验
- 有分布式锁性能优化的经验
- 有分布式锁监控的经验

**架构能力**：
- 能设计高可用的分布式锁方案
- 能设计分布式锁的容灾方案
- 能设计分布式锁的降级方案

**技术选型**：
- 了解 Redis、Zookeeper、Etcd 等多种分布式锁实现
- 能根据业务特点选择合适的方案
- 有自研分布式锁的经验