2025.09.11

黑马点评开发日志Day04——分布式锁

一、分布式锁

1.1 原理及实现方式

  • 分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁
    分布式锁特性

  • 实现方式：常见的有三种

    1. 基于数据库的分布式锁
      • 优点：实现简单，易于理解
      • 缺点：性能较低，数据库压力大，可能导致死锁
    2. 基于缓存的分布式锁（如Redis）
      • 优点：性能较高，适用于高并发场景
      • 缺点：需要处理缓存失效问题，可能导致锁丢失
    3. 基于Zookeeper的分布式锁
      • 优点：高可靠性，适用于复杂分布式系统
      • 缺点：实现复杂，维护成本高
MySQL Redis Zookeeper
互斥 利用mysql本身的互斥锁机制 利用setnx这样的互斥命令 利用节点的唯一性和有序性实现互斥
高可用 好 好 好
高性能 一般 好 一般
安全性 断开连接，自动释放锁 利用锁超时时间，到期释放 临时节点，断开连接自动释放

1.2 基于Redis的分布式锁

  • 实现分布式锁需要实现的两个基本方法
    • 获取锁
      • 互斥：确保只能有一个线程获取锁
        1
        2
        # 添加锁，nx是互斥，ex是设置超时时间
        setnx lock thread1 nx ex 10
    • 释放锁
      • 手动释放
      • 超时释放
        1
        2
        # 释放锁，删除即可
        del lock

实现流程

1.3 基于Redis实现分布式锁初级版本

  • 需求：定义一个类，实现下面接口，利用Redis实现分布式锁功能
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    public interface ILock {
        /**
         * 尝试获取锁
         * @param timeoutSec 锁持有的超时时间，过期后自动释放
         * @return true表示获取成功，false表示获取失败
         */
        boolean tryLock(long timeoutSec);

        /**
         * 释放锁
         */
        void unLock();
    }

1.3.1 SimpleRedisLock

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
public class SimpleRedisLock implements ILock {

    private String name;
    private StringRedisTemplate stringRedisTemplate;
    private static final String KEY_PREFIX = "lock:";

    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取线程标识
        long threadId = Thread.currentThread().getId();

        // 获取锁
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);
        // 防止自动拆箱时出现问题
        return Boolean.TRUE.equals(success);
    }

    @Override
    public void unLock() {
        // 释放锁
        stringRedisTemplate.delete(KEY_PREFIX + name);
    }
}

1.3.2 VoucherOrderServiceImpl

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
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private ISeckillVoucherService seckillVoucherService;

    @Resource
    private RedisIdWorker redisIdWorker;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 优惠券秒杀下单
     * @param voucherId
     * @return
     */
    @Override
    public Result seckillVoucher(Long voucherId) {
        // 1. 查询优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        // 2. 判断秒杀是否开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
            // 尚未开始
            return Result.fail("秒杀尚未开始！");
        }
        // 3. 判断秒杀是否已经结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            // 秒杀已经结束
            return Result.fail("秒杀已经结束");
        }
        // 4. 判断库存是否充足
        if (voucher.getStock() < 1) {
            return Result.fail("库存不足");
        }

        Long userId = UserHolder.getUser().getId();
        // 创建锁对象
        SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        // 获取锁
        boolean isLock = lock.tryLock(5);
        // 判断是否成功
        if (!isLock) {
            // 获取失败，返回错误或重试
            return Result.fail("不允许重复下单");
        }
        try {
            // 获取代理对象
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {
            // 释放锁
            lock.unLock();
        }
    }
}

1.4 Redis分布式锁的误删问题

  • 当线程1阻塞超时时会释放锁，这时线程2获取锁成功，线程1恢复后继续执行删除锁操作，导致线程2的锁被误删
    Redis分布式锁的误删问题
  • 解决方案：给锁添加标识，只有持有锁的线程才能释放锁
    Redis分布式锁的误删问题解决方案
    Redis分布式锁的误删问题解决方案流程

1.5 解决误删问题

  • 需求：修改之前的分布式锁实现，满足：
    • 在获取锁时存入线程标示（可以用UUID表示）
    • 在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一致
      • 如果一致则释放锁
      • 如果不一致则不释放锁

1.5.1 SimpleRedisLock

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
public class SimpleRedisLock implements ILock {

    private String name;
    private StringRedisTemplate stringRedisTemplate;

    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    private static final String KEY_PREFIX = "lock:";
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";


    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取线程标识
        String threadId = ID_PREFIX + Thread.currentThread().getId();

        // 获取锁
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
        // 防止自动拆箱时出现问题
        return Boolean.TRUE.equals(success);
    }

    @Override
    public void unLock() {
        // 获取线程标识
        String threadId = ID_PREFIX + Thread.currentThread().getId();
        // 获取锁中的标识
        String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
        // 判断标识是否一致
        if (threadId.equals(id)) {
            // 释放锁
            stringRedisTemplate.delete(KEY_PREFIX + name);
        }
    }
}

1.6 Lua脚本解决误删问题

  • 上面的解决方案在高并发下会出现问题：线程1现在持有锁之后，在执行业务逻辑过程中，他正准备删除锁，而且已经走到了条件判断的过程中，比如他已经拿到了当前这把锁确实是属于他自己的，正准备删除锁，但是此时他的锁到期了，那么此时线程2进来，但是线程1他会接着往后执行，当他卡顿结束后，他直接就会执行删除锁那行代码，相当于条件判断并没有起到作用，这就是删锁时的原子性问题，之所以有这个问题，是因为线程1的拿锁，比锁，删锁，实际上并不是原子性的，我们要防止刚才的情况发生

Lua脚本解决误删问题

  • Redis的Lua脚本：Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Lua是一种编程语言，它的基本语法大家可以参考网站：https://www.runoob.com/lua/lua-tutorial.html

  • Redis提供的调用函数语法：

    1
    2
    # 执行redis命令
    redis.call('命令名称', 'key', '其它参数', ...)
    • 例如执行set name jack，则脚本是：
    1
    redis.call('set', 'name', 'jack')
    • 例如先执行set name Rose，再执行get name，则脚本是：
    1
    2
    3
    4
    5
    6
    # 先执行set name Rose
    redis.call('set', 'name', 'Rose')
    # 再执行get name
    local name = redis.call('get', 'name')
    # 返回
    return name

  • 调用脚本命令：

    • 无参数
      1
      2
      3
      4
      # 调用脚本
      # "return redis.call('set', 'name', 'jack')"为脚本内容
      # 0表示脚本需要的key类型的参数个数
      EVAL "return redis.call('set', 'name', 'jack')" 0
    • 有参数：如果脚本中的key、value不想写死，可以作为参数传递。key类型参数会放入KEYS数组，其它参数会放入ARGV数组，在脚本中可以从KEYS和ARGV数组获取这些参数
      1
      2
      # 调用脚本
      EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 name jack

  • 释放锁的业务流程是这样的：

    1. 获取锁中的线程标示
    2. 判断是否与指定的标示（当前线程标示）一致
    3. 如果一致则释放锁（删除）
    4. 如果不一致则什么都不做

  • 如果用Lua脚本来表示则是这样的：

    1
    2
    3
    4
    5
    6
    7
    8
    9
    -- KEY[1]表示锁的key
    -- ARGV[1]表示线程标识
    -- 比较线程标识与锁中的标识是否一致
    if (redis.call('get', KEYS[1]) == ARGV[1]) then
        -- 一致，释放锁
        return redis.call('del', KEYS[1])
    end  
    -- 不一致，什么都不做    
    return 0

1.7 基于Lua脚本实现分布式锁的释放逻辑

  • 需求：修改之前的分布式锁实现，利用Lua脚本实现释放锁逻辑
  • 提示：RedisTemplate调用Lua脚本的API如下：
    1
    2
    3
    4
    @Override
    public <T> T execute(RedisScript<T> script, List<K> keys, Object... args) {
      return scriptExecutor.execute(script, keys, args);
    }

1.7.1 unLock.lua

1
2
3
4
5
6
7
8
9
-- KEY[1]表示锁的key
-- ARGV[1]表示线程标识
-- 比较线程标识与锁中的标识是否一致
if (redis.call('get', KEYS[1]) == ARGV[1]) then
    -- 一致，释放锁
    return redis.call('del', KEYS[1])
end
-- 不一致，什么都不做
return 0

1.7.2 SimpleRedisLock

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
public class SimpleRedisLock implements ILock {

    private String name;
    private StringRedisTemplate stringRedisTemplate;

    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    private static final String KEY_PREFIX = "lock:";
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
    private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
    static {
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unLock.lua"));
        UNLOCK_SCRIPT.setResultType(Long.class);
    }

    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取线程标识
        String threadId = ID_PREFIX + Thread.currentThread().getId();

        // 获取锁
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
        // 防止自动拆箱时出现问题
        return Boolean.TRUE.equals(success);
    }

    @Override
    public void unLock() {
        // 调用lua脚本
        stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX + Thread.currentThread().getId());
    }
}

二、Redission

2.1 Redission简介

  • 基于setnx实现的分布式锁存在下面的问题：

    • 不可重入：同一个线程无法多次获取同一把锁
    • 不可重试：获取锁只尝试一次就返回false，没有重试机制
    • 超时释放：锁超时释放虽然可以避免死锁，但如果是业务执行耗时较长，也会导致锁释放，存在安全隐患
    • 主从一致性：如果Redis提供了主从集群，主从同步存在延迟，当主宕机时，如果从并同步主中的锁数据，则会出现锁实现

  • Redisson：是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。

2.2 Redission入门

  • 引入依赖
1
2
3
4
5
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.20.0</version>
</dependency>
  • 配置文件
1
2
3
4
5
6
7
8
9
10
11
12
13
@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient() {
        // 配置
        Config config = new Config();
        // 添加redis地址，这里添加了单点的地址，也可以使用config.useClusterServers()添加集群地址
        config.useSingleServer().setAddress("redis://localhost:6379").setPassword("123456");
        // 创建RedissonClient对象
        return Redisson.create(config);
    }
}
  • 使用RedissonClient获取锁
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
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private ISeckillVoucherService seckillVoucherService;

    @Resource
    private RedisIdWorker redisIdWorker;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private RedissonClient redissonClient;

    /**
     * 优惠券秒杀下单
     * @param voucherId
     * @return
     */
    @Override
    public Result seckillVoucher(Long voucherId) {
        // 1. 查询优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        // 2. 判断秒杀是否开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
            // 尚未开始
            return Result.fail("秒杀尚未开始！");
        }
        // 3. 判断秒杀是否已经结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            // 秒杀已经结束
            return Result.fail("秒杀已经结束");
        }
        // 4. 判断库存是否充足
        if (voucher.getStock() < 1) {
            return Result.fail("库存不足");
        }

        Long userId = UserHolder.getUser().getId();
        // 创建锁对象
        // SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        // 获取锁，参数分别是：获取锁的最大等待时间（期间会重试），锁自动释放时间，时间单位
        boolean isLock = lock.tryLock(1, 5, TimeUnit.SECONDS);
        // 判断是否成功
        if (!isLock) {
            // 获取失败，返回错误或重试
            return Result.fail("不允许重复下单");
        }
        try {
            // 获取代理对象
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {
            // 释放锁
            lock.unlock();
        }
    }
}

2.3 Redission可重入锁原理

  • 使用hash存储锁
    • key：锁的名称
    • field：线程标识
    • value：重入次数，获取成功时+1，释放锁时-1，减到0时删除锁

Redission可重入锁原理

  • 获取锁的Lua脚本
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
local key = KEYS[1]; -- 锁的key
local threadId = ARGV[1]; -- 线程唯一标识
local releaseTime = ARGV[2]; -- 锁的自动释放时间
-- 判断是否存在
if(redis.call('exists', key) == 0) then
    -- 不存在, 获取锁    
    redis.call('hset', key, threadId, '1'); 
    -- 设置有效期    
    redis.call('expire', key, releaseTime);     
    return 1; -- 返回结果
end;
-- 锁已经存在，判断threadId是否是自己
if(redis.call('hexists', key, threadId) == 1) then
    -- 不存在, 获取锁，重入次数+1    
    redis.call('hincrby', key, threadId, '1'); 
    -- 设置有效期
    redis.call('expire', key, releaseTime); 
    return 1; -- 返回结果
end;
return 0; -- 代码走到这里,说明获取锁的不是自己，获取锁失败
  • 释放锁的Lua脚本
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
local key = KEYS[1]; -- 锁的key
local threadId = ARGV[1]; -- 线程唯一标识
local releaseTime = ARGV[2]; -- 锁的自动释放时间
-- 判断当前锁是否还是被自己持有
if (redis.call('HEXISTS', key, threadId) == 0) then    
    return nil; -- 如果已经不是自己，则直接返回
end;
-- 是自己的锁，则重入次数-1
local count = redis.call('HINCRBY', key, threadId, -1);
-- 判断是否重入次数是否已经为0 
if (count > 0) then    
    -- 大于0说明不能释放锁，重置有效期然后返回    
    redis.call('EXPIRE', key, releaseTime);    
    return nil;
else  -- 等于0说明可以释放锁，直接删除    
    redis.call('DEL', key);    
    return nil;
end;

2.4 Redission的锁重试和WacthDog机制

  • 可重试：利用信号量和PubSub功能实现等待、唤醒，获取锁失败的重试机制
  • 超时续约：利用watchDog，每隔一段时间（releaseTime / 3），重置超时时间

Redission的锁重试和WacthDog机制

2.5 Redission的multiLock机制

  • Redission的multiLock可以将多把锁合成一把锁，获取这把锁时需要同时获取所有的锁，释放这把锁时会同时释放所有的锁

Redission的multiLock机制

2.6 总结

  1. 不可重入Redis分布式锁：
    • 原理：利用setnx的互斥性；利用ex避免死锁；释放锁时判断线程标示
    • 缺陷：不可重入、无法重试、锁超时失效
  2. 可重入的Redis分布式锁：
    • 原理：利用hash结构，记录线程标示和重入次数；利用watchDog延续锁时间；利用信号量控制锁重试等待
    • 缺陷：redis宕机引起锁失效问题
  3. Redisson的multiLock：
    • 原理：多个独立的Redis节点，必须在所有节点都获取重入锁，才算获取锁成功
    • 缺陷：运维成本高、实现复杂