分布式锁的正确用法与误区？

1. 概述

1.1 分布式锁的重要性

分布式锁是分布式系统设计的核心组件之一，用于在分布式环境下保证资源的互斥访问。

分布式锁的应用场景：

  • 防止重复操作：防止重复支付、重复下单等
  • 资源互斥访问：保证同一时间只有一个请求能访问共享资源
  • 分布式任务调度：保证定时任务只在一个节点执行
  • 缓存更新：防止缓存击穿、缓存雪崩

1.2 分布式锁的挑战

分布式锁的挑战：

  • 死锁问题：锁未正确释放导致死锁
  • 锁超时问题：锁超时时间设置不当
  • 锁续期问题：长时间任务需要锁续期
  • 性能问题：锁成为性能瓶颈
  • 可用性问题：锁服务故障导致系统不可用

1.3 本文内容结构

本文将从以下几个方面全面解析分布式锁：

  1. 分布式锁原理：什么是分布式锁、为什么需要分布式锁
  2. 实现方式：Redis、ZooKeeper、数据库等实现方式
  3. 正确用法：正确的使用方式和最佳实践
  4. 常见误区：常见错误和如何避免
  5. 实战案例：实际项目中的分布式锁使用

2. 分布式锁原理

2.1 什么是分布式锁

2.1.1 定义

分布式锁：在分布式环境下，用于保证同一时间只有一个进程或线程能访问共享资源的机制。

特点：

  • 互斥性：同一时间只有一个进程能持有锁
  • 可重入性：同一进程可以多次获取锁
  • 锁超时：锁有超时时间，防止死锁
  • 高可用：锁服务需要高可用

2.1.2 与本地锁的区别

本地锁（synchronized、ReentrantLock）：

  • 只作用于单个JVM进程
  • 不能跨进程、跨机器
  • 性能高，无网络开销

分布式锁：

  • 作用于多个JVM进程
  • 可以跨进程、跨机器
  • 性能较低，有网络开销

2.2 为什么需要分布式锁

2.2.1 分布式环境下的问题

问题场景：

  • 多实例部署：应用部署在多个节点
  • 共享资源：多个节点访问共享资源（数据库、缓存等）
  • 并发访问：多个请求同时访问共享资源

示例：

  • 用户下单，多个实例同时处理，可能导致重复下单
  • 定时任务，多个实例同时执行，可能导致重复执行

2.2.2 分布式锁的作用

分布式锁的作用：

  • 保证互斥：同一时间只有一个请求能执行
  • 防止重复：防止重复操作
  • 保证一致性：保证数据一致性

3. 分布式锁实现方式

3.1 Redis分布式锁

3.1.1 基本原理

Redis分布式锁：使用Redis的SET命令实现分布式锁。

实现原理：

  • 使用SET key value NX EX timeout命令
  • NX：只在键不存在时设置
  • EX：设置过期时间
  • 释放锁时删除键

3.1.2 基础实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@Service
public class RedisDistributedLock {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    /**
     * 获取锁
     */
    public boolean tryLock(String lockKey, String lockValue, long expireTime, TimeUnit timeUnit) {
        Boolean result = redisTemplate.opsForValue().setIfAbsent(
            lockKey,
            lockValue,
            expireTime,
            timeUnit
        );
        return Boolean.TRUE.equals(result);
    }
    
    /**
     * 释放锁
     */
    public void unlock(String lockKey, String lockValue) {
        // 使用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<>();
        script.setScriptText(luaScript);
        script.setResultType(Long.class);
        
        redisTemplate.execute(script, Collections.singletonList(lockKey), lockValue);
    }
}

3.1.3 Redisson实现（推荐）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@Service
public class RedissonDistributedLock {
    
    @Autowired
    private RedissonClient redissonClient;
    
    /**
     * 获取锁（推荐使用Redisson）
     */
    public boolean tryLock(String lockKey, long waitTime, long leaseTime, TimeUnit timeUnit) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(waitTime, leaseTime, timeUnit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
    
    /**
     * 释放锁
     */
    public void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        if (lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }
    
    /**
     * 使用示例
     */
    public void doSomething(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            // 尝试加锁，最多等待100ms，锁定30秒
            if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
                try {
                    // 执行业务逻辑
                    doBusinessLogic();
                } finally {
                    // 释放锁
                    lock.unlock();
                }
            } else {
                throw new BusinessException("获取锁失败");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("获取锁被中断", e);
        }
    }
}

3.2 ZooKeeper分布式锁

3.2.1 基本原理

ZooKeeper分布式锁：使用ZooKeeper的临时顺序节点实现分布式锁。

实现原理：

  • 创建临时顺序节点
  • 获取所有子节点，判断自己是否是最小节点
  • 如果是，获取锁；否则，监听前一个节点
  • 释放锁时删除节点

3.2.2 实现代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@Service
public class ZooKeeperDistributedLock {
    
    @Autowired
    private CuratorFramework curatorFramework;
    
    private static final String LOCK_PATH = "/distributed-lock";
    
    /**
     * 获取锁
     */
    public InterProcessMutex acquireLock(String lockKey) {
        String lockPath = LOCK_PATH + "/" + lockKey;
        InterProcessMutex mutex = new InterProcessMutex(curatorFramework, lockPath);
        return mutex;
    }
    
    /**
     * 使用示例
     */
    public void doSomething(String lockKey) {
        InterProcessMutex mutex = acquireLock(lockKey);
        try {
            // 获取锁，最多等待10秒
            if (mutex.acquire(10, TimeUnit.SECONDS)) {
                try {
                    // 执行业务逻辑
                    doBusinessLogic();
                } finally {
                    // 释放锁
                    mutex.release();
                }
            } else {
                throw new BusinessException("获取锁失败");
            }
        } catch (Exception e) {
            throw new BusinessException("获取锁异常", e);
        }
    }
}

3.3 数据库分布式锁

3.3.1 基本原理

数据库分布式锁：使用数据库的唯一索引或行锁实现分布式锁。

实现方式：

  • 唯一索引：插入唯一记录作为锁
  • 行锁：使用SELECT FOR UPDATE

3.3.2 实现代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@Service
public class DatabaseDistributedLock {
    
    @Autowired
    private DistributedLockMapper lockMapper;
    
    /**
     * 获取锁（唯一索引方式）
     */
    public boolean tryLock(String lockKey, String lockValue, long expireTime) {
        try {
            DistributedLock lock = new DistributedLock();
            lock.setLockKey(lockKey);
            lock.setLockValue(lockValue);
            lock.setExpireTime(LocalDateTime.now().plusSeconds(expireTime));
            lockMapper.insert(lock);
            return true;
        } catch (DuplicateKeyException e) {
            // 锁已存在，检查是否过期
            DistributedLock existingLock = lockMapper.selectByLockKey(lockKey);
            if (existingLock != null && existingLock.getExpireTime().isBefore(LocalDateTime.now())) {
                // 锁已过期，删除并重新获取
                lockMapper.deleteByLockKey(lockKey);
                return tryLock(lockKey, lockValue, expireTime);
            }
            return false;
        }
    }
    
    /**
     * 释放锁
     */
    public void unlock(String lockKey, String lockValue) {
        DistributedLock lock = lockMapper.selectByLockKey(lockKey);
        if (lock != null && lock.getLockValue().equals(lockValue)) {
            lockMapper.deleteByLockKey(lockKey);
        }
    }
    
    /**
     * 获取锁（行锁方式）
     */
    @Transactional
    public boolean tryLockWithRowLock(String lockKey) {
        // 使用SELECT FOR UPDATE获取行锁
        DistributedLock lock = lockMapper.selectForUpdate(lockKey);
        if (lock == null) {
            // 锁不存在，创建锁
            lock = new DistributedLock();
            lock.setLockKey(lockKey);
            lock.setExpireTime(LocalDateTime.now().plusSeconds(30));
            lockMapper.insert(lock);
            return true;
        } else if (lock.getExpireTime().isBefore(LocalDateTime.now())) {
            // 锁已过期，更新锁
            lock.setExpireTime(LocalDateTime.now().plusSeconds(30));
            lockMapper.updateById(lock);
            return true;
        }
        return false;
    }
}

4. 分布式锁的正确用法

4.1 基本使用模式

4.1.1 标准模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Service
public class OrderService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    /**
     * 标准使用模式
     */
    public void createOrder(OrderRequest request) {
        String lockKey = "order:lock:" + request.getUserId();
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 1. 尝试加锁
            if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
                try {
                    // 2. 执行业务逻辑
                    doCreateOrder(request);
                } finally {
                    // 3. 释放锁（必须在finally中释放）
                    lock.unlock();
                }
            } else {
                throw new BusinessException("系统繁忙，请稍后再试");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("获取锁被中断", e);
        }
    }
}

4.1.2 关键要点

关键要点：

  1. 锁必须在finally中释放：确保锁一定会被释放
  2. 检查锁的持有者：释放锁前检查是否是当前线程持有
  3. 设置合理的超时时间：避免死锁
  4. 处理中断异常：正确处理InterruptedException

4.2 锁超时时间设置

4.2.1 超时时间的重要性

超时时间的作用：

  • 防止死锁：锁超时后自动释放
  • 保证可用性：避免锁一直占用资源

设置原则：

  • 不能太短：业务逻辑未执行完就超时
  • 不能太长：死锁时等待时间过长

4.2.2 动态超时时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@Service
public class OrderService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    /**
     * 动态超时时间
     */
    public void createOrder(OrderRequest request) {
        String lockKey = "order:lock:" + request.getUserId();
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 1. 先获取锁，不设置超时时间
            if (lock.tryLock(100, TimeUnit.MILLISECONDS)) {
                try {
                    // 2. 执行业务逻辑
                    long startTime = System.currentTimeMillis();
                    doCreateOrder(request);
                    long duration = System.currentTimeMillis() - startTime;
                    
                    // 3. 根据实际执行时间设置锁超时时间
                    // 锁超时时间 = 执行时间 * 2（预留缓冲）
                    long lockTimeout = Math.max(duration * 2, 30000); // 最少30秒
                    lock.expire(lockTimeout, TimeUnit.MILLISECONDS);
                } finally {
                    lock.unlock();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("获取锁被中断", e);
        }
    }
}

4.3 锁续期（Watch Dog）

4.3.1 锁续期的必要性

锁续期的场景：

  • 长时间任务：业务逻辑执行时间超过锁超时时间
  • 不确定执行时间：无法预估业务逻辑执行时间

Redisson的Watch Dog机制：

  • 自动续期：锁超时时间的三分之一时自动续期
  • 默认续期时间：30秒

4.3.2 使用Watch Dog

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@Service
public class OrderService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    /**
     * 使用Watch Dog自动续期
     */
    public void createOrder(OrderRequest request) {
        String lockKey = "order:lock:" + request.getUserId();
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 1. 获取锁，不设置leaseTime，使用Watch Dog自动续期
            if (lock.tryLock(100, TimeUnit.MILLISECONDS)) {
                try {
                    // 2. 执行业务逻辑（Watch Dog会自动续期）
                    doCreateOrder(request);
                } finally {
                    // 3. 释放锁（Watch Dog会自动停止）
                    lock.unlock();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("获取锁被中断", e);
        }
    }
}

4.4 可重入锁

4.4.1 可重入锁的必要性

可重入锁的场景：

  • 递归调用：方法内部递归调用
  • 方法调用链：方法A调用方法B，都需要获取锁

Redisson的RLock：

  • 支持可重入
  • 记录重入次数
  • 释放锁时减少重入次数

4.4.2 使用可重入锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@Service
public class OrderService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    /**
     * 可重入锁示例
     */
    public void createOrder(OrderRequest request) {
        String lockKey = "order:lock:" + request.getUserId();
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
                try {
                    // 调用其他方法，也需要获取锁（可重入）
                    validateOrder(request);
                    processOrder(request);
                } finally {
                    lock.unlock();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("获取锁被中断", e);
        }
    }
    
    /**
     * 内部方法也需要获取锁（可重入）
     */
    private void validateOrder(OrderRequest request) {
        String lockKey = "order:lock:" + request.getUserId();
        RLock lock = redissonClient.getLock(lockKey);
        
        // 可重入：同一个线程可以再次获取锁
        if (lock.tryLock()) {
            try {
                // 验证订单逻辑
            } finally {
                lock.unlock();
            }
        }
    }
}

4.5 读写锁

4.5.1 读写锁的应用场景

读写锁的场景：

  • 读多写少：读操作可以并发，写操作需要互斥
  • 缓存更新：多个读操作可以并发，写操作需要互斥

4.5.2 使用读写锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@Service
public class CacheService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    /**
     * 读操作（使用读锁）
     */
    public String getCache(String key) {
        String lockKey = "cache:lock:" + key;
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(lockKey);
        RLock readLock = readWriteLock.readLock();
        
        try {
            if (readLock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
                try {
                    // 读操作可以并发执行
                    return getCacheFromRedis(key);
                } finally {
                    readLock.unlock();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return null;
    }
    
    /**
     * 写操作（使用写锁）
     */
    public void updateCache(String key, String value) {
        String lockKey = "cache:lock:" + key;
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(lockKey);
        RLock writeLock = readWriteLock.writeLock();
        
        try {
            if (writeLock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
                try {
                    // 写操作需要互斥执行
                    updateCacheToRedis(key, value);
                } finally {
                    writeLock.unlock();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

5. 分布式锁的常见误区

5.1 误区1：锁未在finally中释放

5.1.1 错误示例

1
2
3
4
5
6
7
8
// 错误示例：锁未在finally中释放
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    if (lock.tryLock()) {
        doCreateOrder(request);
        lock.unlock(); // 如果这里抛异常，锁不会被释放
    }
}

5.1.2 正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 正确示例：锁在finally中释放
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    try {
        if (lock.tryLock()) {
            try {
                doCreateOrder(request);
            } finally {
                lock.unlock(); // 确保锁一定会被释放
            }
        }
    } catch (Exception e) {
        // 处理异常
    }
}

5.2 误区2：释放了其他线程的锁

5.2.1 错误示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 错误示例：可能释放其他线程的锁
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    try {
        if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
            try {
                doCreateOrder(request);
            } finally {
                // 如果锁超时被其他线程获取，这里会释放其他线程的锁
                lock.unlock();
            }
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

5.2.2 正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 正确示例：检查锁的持有者
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    try {
        if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
            try {
                doCreateOrder(request);
            } finally {
                // 检查是否是当前线程持有锁
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

5.3 误区3：锁超时时间设置不当

5.3.1 错误示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 错误示例：锁超时时间太短
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    try {
        // 锁超时时间只有1秒，业务逻辑可能需要10秒
        if (lock.tryLock(100, 1, TimeUnit.SECONDS)) {
            try {
                doCreateOrder(request); // 可能执行10秒
                // 锁在1秒后超时，其他线程可以获取锁，导致并发问题
            } finally {
                lock.unlock();
            }
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

5.3.2 正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// 正确示例：根据业务逻辑设置合理的超时时间
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    try {
        // 锁超时时间设置为业务逻辑预估时间的2倍
        long estimatedTime = estimateBusinessTime(request);
        long lockTimeout = Math.max(estimatedTime * 2, 30000); // 最少30秒
        
        if (lock.tryLock(100, lockTimeout, TimeUnit.MILLISECONDS)) {
            try {
                doCreateOrder(request);
            } finally {
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

5.4 误区4：未处理锁获取失败

5.4.1 错误示例

1
2
3
4
5
6
7
8
9
10
// 错误示例：未处理锁获取失败
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    lock.lock(); // 如果获取锁失败，会一直阻塞
    try {
        doCreateOrder(request);
    } finally {
        lock.unlock();
    }
}

5.4.2 正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 正确示例：使用tryLock，设置超时时间
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    try {
        // 尝试获取锁，最多等待100ms
        if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
            try {
                doCreateOrder(request);
            } finally {
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        } else {
            // 获取锁失败，返回错误或重试
            throw new BusinessException("系统繁忙，请稍后再试");
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new BusinessException("获取锁被中断", e);
    }
}

5.5 误区5：锁粒度太粗

5.5.1 错误示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 错误示例：锁粒度太粗，影响性能
public void createOrder(OrderRequest request) {
    // 所有订单共用一个锁，性能差
    RLock lock = redissonClient.getLock("order:lock");
    try {
        if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
            try {
                doCreateOrder(request);
            } finally {
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

5.5.2 正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 正确示例：锁粒度细化，按用户ID加锁
public void createOrder(OrderRequest request) {
    // 每个用户一个锁，不同用户可以并发
    String lockKey = "order:lock:" + request.getUserId();
    RLock lock = redissonClient.getLock(lockKey);
    try {
        if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
            try {
                doCreateOrder(request);
            } finally {
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

5.6 误区6：未考虑锁服务故障

5.6.1 错误示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 错误示例：未考虑Redis故障
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    try {
        if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
            try {
                doCreateOrder(request);
            } finally {
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock(); // 如果Redis故障，这里会抛异常
                }
            }
        }
    } catch (Exception e) {
        // 未处理Redis故障
    }
}

5.6.2 正确做法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
// 正确示例：处理锁服务故障，提供降级方案
public void createOrder(OrderRequest request) {
    RLock lock = redissonClient.getLock("order:lock");
    try {
        if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
            try {
                doCreateOrder(request);
            } finally {
                try {
                    if (lock.isHeldByCurrentThread()) {
                        lock.unlock();
                    }
                } catch (Exception e) {
                    // Redis故障，记录日志，但不影响业务
                    log.error("释放锁失败", e);
                    // 可以考虑发送告警
                }
            }
        }
    } catch (Exception e) {
        // Redis故障，提供降级方案
        if (e instanceof RedisException) {
            // 降级：使用数据库唯一索引保证幂等性
            return createOrderWithDatabaseLock(request);
        }
        throw e;
    }
}

6. 实战案例

6.1 案例1：防止重复下单

6.1.1 场景

需求：防止用户重复下单，高并发场景。

解决方案：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@Service
public class OrderService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    @Autowired
    private OrderMapper orderMapper;
    
    /**
     * 创建订单（使用分布式锁防止重复下单）
     */
    public OrderResult createOrder(OrderRequest request) {
        // 1. 锁粒度：按用户ID加锁
        String lockKey = "order:create:lock:" + request.getUserId();
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 2. 尝试获取锁，最多等待100ms，锁定30秒
            if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
                try {
                    // 3. 双重检查：检查是否已下单
                    Order existingOrder = orderMapper.selectByUserIdAndSkuId(
                        request.getUserId(),
                        request.getSkuId()
                    );
                    if (existingOrder != null) {
                        return OrderResult.success(existingOrder, "订单已存在");
                    }
                    
                    // 4. 创建订单
                    Order order = new Order();
                    order.setUserId(request.getUserId());
                    order.setSkuId(request.getSkuId());
                    order.setQuantity(request.getQuantity());
                    order.setAmount(request.getAmount());
                    order.setStatus(OrderStatus.PENDING);
                    orderMapper.insert(order);
                    
                    return OrderResult.success(order);
                } finally {
                    // 5. 释放锁
                    if (lock.isHeldByCurrentThread()) {
                        lock.unlock();
                    }
                }
            } else {
                throw new BusinessException("系统繁忙，请稍后再试");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("获取锁被中断", e);
        }
    }
}

6.2 案例2：防止重复支付

6.2.1 场景

需求：防止订单重复支付，保证支付幂等性。

解决方案：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@Service
public class PaymentService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    @Autowired
    private PaymentRecordMapper paymentRecordMapper;
    
    @Autowired
    private OrderMapper orderMapper;
    
    /**
     * 支付订单（使用分布式锁防止重复支付）
     */
    @Transactional
    public PaymentResult payOrder(Long orderId, BigDecimal amount) {
        // 1. 锁粒度：按订单ID加锁
        String lockKey = "payment:lock:" + orderId;
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
                try {
                    // 2. 检查订单状态
                    Order order = orderMapper.selectById(orderId);
                    if (order == null) {
                        throw new BusinessException("订单不存在");
                    }
                    
                    // 3. 检查是否已支付（幂等性检查）
                    if (order.getStatus() == OrderStatus.PAID) {
                        PaymentRecord record = paymentRecordMapper.selectByOrderId(orderId);
                        return PaymentResult.success(record, "订单已支付");
                    }
                    
                    // 4. 检查订单状态
                    if (order.getStatus() != OrderStatus.PENDING) {
                        throw new BusinessException("订单状态不允许支付");
                    }
                    
                    // 5. 检查金额
                    if (order.getAmount().compareTo(amount) != 0) {
                        throw new BusinessException("支付金额不匹配");
                    }
                    
                    // 6. 创建支付记录
                    PaymentRecord record = new PaymentRecord();
                    record.setOrderId(orderId);
                    record.setAmount(amount);
                    record.setStatus(PaymentStatus.SUCCESS);
                    paymentRecordMapper.insert(record);
                    
                    // 7. 更新订单状态（乐观锁）
                    int updated = orderMapper.updateStatusWithVersion(
                        orderId,
                        OrderStatus.PENDING,
                        OrderStatus.PAID,
                        order.getVersion()
                    );
                    
                    if (updated == 0) {
                        // 状态已变更，可能是并发支付
                        order = orderMapper.selectById(orderId);
                        if (order.getStatus() == OrderStatus.PAID) {
                            record = paymentRecordMapper.selectByOrderId(orderId);
                            return PaymentResult.success(record, "订单已支付");
                        }
                        throw new BusinessException("订单状态已变更");
                    }
                    
                    return PaymentResult.success(record);
                } finally {
                    if (lock.isHeldByCurrentThread()) {
                        lock.unlock();
                    }
                }
            } else {
                throw new BusinessException("系统繁忙，请稍后再试");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("获取锁被中断", e);
        }
    }
}

6.3 案例3：分布式定时任务

6.3.1 场景

需求：定时任务在多个实例中只执行一次。

解决方案：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@Component
public class ScheduledTask {
    
    @Autowired
    private RedissonClient redissonClient;
    
    /**
     * 定时任务（使用分布式锁保证只执行一次）
     */
    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点执行
    public void executeTask() {
        String lockKey = "scheduled:task:lock";
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 尝试获取锁，不等待，锁定1小时
            if (lock.tryLock(0, 1, TimeUnit.HOURS)) {
                try {
                    // 执行定时任务
                    doScheduledTask();
                } finally {
                    if (lock.isHeldByCurrentThread()) {
                        lock.unlock();
                    }
                }
            } else {
                // 其他实例已执行，跳过
                log.info("Task already executed by another instance");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    
    private void doScheduledTask() {
        // 定时任务逻辑
        log.info("Executing scheduled task");
    }
}

7. 性能优化

7.1 锁粒度优化

7.1.1 细粒度锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Service
public class OrderService {
    
    /**
     * 锁粒度优化：按用户ID加锁，而不是全局锁
     */
    public void createOrder(OrderRequest request) {
        // 细粒度：每个用户一个锁
        String lockKey = "order:lock:" + request.getUserId();
        RLock lock = redissonClient.getLock(lockKey);
        
        // 而不是：所有用户共用一个锁
        // String lockKey = "order:lock";
    }
}

7.2 分段锁

7.2.1 分段锁实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@Service
public class AccountService {
    
    /**
     * 分段锁：将锁分散到多个分段，减少锁竞争
     */
    public void transfer(TransferRequest request) {
        // 根据账户ID取模，将锁分散到100个分段
        int segment = (int) (request.getFromAccountId() % 100);
        String lockKey = "transfer:lock:segment:" + segment;
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            if (lock.tryLock(100, 30, TimeUnit.MILLISECONDS)) {
                try {
                    doTransfer(request);
                } finally {
                    if (lock.isHeldByCurrentThread()) {
                        lock.unlock();
                    }
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

7.3 锁超时优化

7.3.1 动态超时时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Service
public class OrderService {
    
    /**
     * 动态超时时间：根据业务逻辑执行时间动态调整
     */
    public void createOrder(OrderRequest request) {
        String lockKey = "order:lock:" + request.getUserId();
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 先获取锁，不设置超时时间
            if (lock.tryLock(100, TimeUnit.MILLISECONDS)) {
                try {
                    long startTime = System.currentTimeMillis();
                    doCreateOrder(request);
                    long duration = System.currentTimeMillis() - startTime;
                    
                    // 根据实际执行时间设置超时时间
                    long lockTimeout = Math.max(duration * 2, 30000);
                    lock.expire(lockTimeout, TimeUnit.MILLISECONDS);
                } finally {
                    if (lock.isHeldByCurrentThread()) {
                        lock.unlock();
                    }
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

8. 总结

8.1 核心要点

  1. 分布式锁实现方式：Redis、ZooKeeper、数据库
  2. 正确用法：锁在finally中释放、检查锁持有者、设置合理超时时间、使用Watch Dog
  3. 常见误区：锁未释放、释放其他线程的锁、超时时间不当、未处理获取失败、锁粒度太粗、未考虑锁服务故障
  4. 性能优化：细粒度锁、分段锁、动态超时时间
  5. 最佳实践：根据场景选择合适的实现方式，遵循正确的使用模式

8.2 关键理解

  1. 锁必须在finally中释放：确保锁一定会被释放
  2. 检查锁的持有者：防止释放其他线程的锁
  3. 设置合理的超时时间：防止死锁，保证可用性
  4. 锁粒度要合适：不能太粗，也不能太细
  5. 考虑锁服务故障：提供降级方案

8.3 最佳实践

  1. 推荐使用Redisson：功能完善，支持Watch Dog、可重入锁、读写锁
  2. 锁粒度细化：按业务维度加锁，而不是全局锁
  3. 使用tryLock：设置超时时间，避免无限等待
  4. 处理异常：正确处理InterruptedException和锁服务故障
  5. 性能优化：使用分段锁、动态超时时间等优化手段

相关文章：