数据库
缓存数据库 - Redis

Redis - 分布式锁

笔记

当多个 Redis 部署在多个服务器上，形成集群，那么就需要实现分布式锁，来控制共享资源的访问，提高效率。

1. 问题描述
2. 分布式锁指令
- Redis的分布式锁实现(推荐)
- 早期Redis的分布式锁实现
3. Java分布式锁流程
4. 总结

# 1. 问题描述

随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的 Java API 并不能提供分布式锁的能力。为了解决这个问题就需要一种跨 JVM 的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题。

分布式锁的实现通常有以下几种方案：

基于数据库实现分布式锁：通过数据库表记录锁的信息（如锁的持有者、锁的状态等），利用数据库提供的原子性操作来实现锁的获取和释放。这种方式简单易实施，但性能较低，且在数据库宕机的情况下可用性不高。
基于缓存（如Redis）实现分布式锁：利用Redis等缓存系统的原子命令和过期机制实现锁的互斥访问。性能较高，适用于对响应时间要求严格的场景。但在极端情况下可能会出现锁丢失的情况。
基于ZooKeeper实现分布式锁：通过ZooKeeper的临时顺序节点来实现锁的互斥访问，具有非常高的可靠性和一致性。但其性能相对较低，且使用相对复杂。

方案比较

性能方面：Redis方案提供了最高的性能，适用于对延迟和吞吐量有较高要求的场景。
可靠性方面：ZooKeeper方案在可靠性方面表现最优，其严格的顺序一致性保证了锁的安全性和一致性。
易用性方面：数据库方案在易

本内容，我们就基于 Redis 实现分布式锁。

# 2. 分布式锁指令

Redis分布式锁是一种在分布式系统中保证资源互斥访问的机制，通过Redis提供的特定指令实现。这种锁机制常用于处理分布式系统中的并发操作，确保同一时刻只有一个客户端能操作特定资源。

# Redis的分布式锁实现(推荐)

Redis提供了SET指令的扩展参数，用于实现更灵活的设置操作，这对于实现分布式锁特别有用。具体使用方法如下：

SET <key> <value> [NX|XX] [PX|EX] <time>

<key>和<value>：分别表示锁的唯一标识和锁的值。
NX和XX：NX表示键不存在时才设置键，用于加锁操作；XX表示键已经存在时才设置键，用于锁的续期操作。
PX和EX：设置键的过期时间，PX表示过期时间单位为毫秒，EX表示过期时间单位为秒。

加锁：为了加锁，使用NX参数确保只有在键不存在时才设置锁，即只有第一个尝试加锁的客户端能成功设置锁：

SET lock_key unique_value NX PX 30000

这里，lock_key是锁的键，unique_value是客户端生成的唯一值（用于解锁时验证锁的所有者），30000表示锁的过期时间是30000毫秒。

解锁：解锁时需要验证unique_value，确保只有锁的持有者才能释放锁，通常通过Lua脚本来原子地执行这一操作：

if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end

1
2
3
4
5

这个Lua脚本比较键lock_key的值和提供的unique_value，如果匹配，则删除键以释放锁；否则，不做任何操作。

# 早期Redis的分布式锁实现

在Redis的早期版本中，实现分布式锁主要依赖于SETNX和SETEX这两个指令，它们各自承担了锁的创建和锁的超时设置这两个关键步骤。

1. 使用SETNX实现锁的互斥性

SETNX指令的全称是"SET if Not eXists"，其作用是只有当指定的键不存在时，才会将键值对设置到数据库中。这一特性使得SETNX非常适合用于实现锁的互斥性。具体使用方式如下：

SETNX lock_key unique_value

lock_key是锁的键，用于标识锁。
unique_value是锁的值，通常可以是一个随机生成的唯一值，用于标识锁的所有者。

如果SETNX操作返回1，表示成功获取锁；如果返回0，则表示锁已经存在，即其他客户端已经持有了锁。

释放锁的操作在早期版本的Redis中没有专门的指令来直接完成。通常，释放锁的逻辑是直接删除对应的键：

DEL lock_key

这里lock_key是在加锁时使用SETNX或SETEX命令设置的键。通过删除这个键，锁被释放，其他客户端就可以再次尝试加锁。

早期分布式锁的指令 `setnx`使用总结，该指令的功能是：

如果插入的 key 没有存在 Redis，则将 key-value 存入 Redis
如果插入的 key 已经存在 Redis，则 value 失效，无法重新覆盖原来的 value

这样就实现了分布式锁：key 存在则代表有人操作，其他人无法操作。

2. 使用SETEX设置锁的超时时间

为了避免死锁的发生，锁需要有一个超时时间。这可以通过SETEX指令来实现。SETEX是"SET with EXpiry"的缩写，它允许在设置键值的同时设置键的过期时间。其使用方式如下：

SETEX lock_key timeout unique_value

timeout是锁的超时时间，单位是秒。这个参数确保了即使锁的持有者因为某些原因未能释放锁，锁也会在超时后自动被Redis删除，从而允许其他客户端重新竞争获取锁。

3. 早期版本的局限性

虽然通过SETNX和SETEX的组合使用可以实现带有超时时间的分布式锁，但这种方法存在一个显著的局限性：设置锁和设置锁的超时时间是两个独立的操作，它们之间无法保证原子性。如果在SETNX和SETEX之间的操作中断，可能会导致锁永远不会过期，从而引发死锁。

为了解决这个问题，较新版本的Redis提供了SET指令的扩展用法，允许在单个操作中同时设置键值对、设置互斥和设置超时时间，确保了操作的原子性，成为了实现分布式锁的推荐方法。

# 3. Java分布式锁流程

拿锁
业务操作
释放锁

properties 配置文件内容：

server.port=8080

# Redis 服务器地址
spring.redis.host=192.168.199.27
# Redis 服务器连接端口
spring.redis.port=6379
# Redis 数据库索引（默认为 0）
spring.redis.database= 0
# 连接超时时间（毫秒）
spring.redis.timeout=1800000
# 连接池最大连接数（使用负值表示没有限制）
spring.redis.lettuce.pool.max-active=20
# 最大阻塞等待时间(负数表示没限制)
spring.redis.lettuce.pool.max-wait=-1
# 连接池中的最大空闲连接
spring.redis.lettuce.pool.max-idle=5
# 连接池中的最小空闲连接
spring.redis.lettuce.pool.min-idle=0

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

Redis 核心配置类：

@EnableCaching
@Configuration
public class RedisConfig extends CachingConfigurerSupport {
   
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new
                Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        template.setConnectionFactory(factory);
        //key 序列化方式
        template.setKeySerializer(redisSerializer);
        //value 序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //value hashmap 序列化
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        return template;
    }

    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory) {
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        //解决查询缓存转换异常的问题
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // 配置序列化（解决乱码的问题）,过期时间 600 秒
        RedisCacheConfiguration config =
                RedisCacheConfiguration.defaultCacheConfig()
                        .entryTtl(Duration.ofSeconds(600))
                        .serializeKeysWith(RedisSerializationContext.SerializationPair.
                                fromSerializer(redisSerializer))
                        .serializeValuesWith(RedisSerializationContext.SerializationPair
                                .fromSerializer(jackson2JsonRedisSerializer))
                        .disableCachingNullValues();
        RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
                .cacheDefaults(config)
                .build();
        return cacheManager;
    }
}

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

# 代码一（无过期时间）

public class RedisLocked {
    @Autowired
    private RedisTemplate<String,String> redisTemplate;
    
    @GetMapping("testLock1")
    public void testLock(){
        // 1 获取锁，setne
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111");
        // 2 获取锁成功、查询 num 的值
        if(lock){
            Object value = redisTemplate.opsForValue().get("num");
            // 2.1 判断 num 为空 return
            if(StringUtils.isEmpty(value)){
                return;
            }
            // 2.2 有值就转成成 int
            int num = Integer.parseInt(value + "");
            // 2.3 把 redis 的 num 加 1
            redisTemplate.opsForValue().set("num", String.valueOf(++num));
            // 2.4 释放锁，del
            redisTemplate.delete("lock");
        }else{
            // 3 获取锁失败、每隔 0.1 秒再获取
            try {
                Thread.sleep(100);
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

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

重启 Redis 服务集群，这里利用 ab 网关压力测试：（已经在 Redis - 事务与锁讲解安装和使用）

ab -n 1000 -c 100 http://192.168.199.1:8080/test/testLock

192.168.199.1 是本机的 IP，此时是 Linux 系统访问本机的 Spring Boot 项目。

查看 redis 中 num 的值：

可能出现的问题：setnx 刚好获取到锁，业务逻辑出现异常 Exception，导致锁无法释放，卡死。

解决：设置过期时间，自动释放锁。

# 优化之设置锁的过期时间

设置过期时间有两种方式：

首先想到通过 expire 设置过期时间（缺乏原子性：如果在 setnx 和 expire 之间出现异常，锁也无法释放）
在 set 时指定过期时间（推荐）

# 代码二（无唯一标识）

在代码一的基础上加上超时时间，看第八行代码

public class RedisLocked {
    @Autowired
    private RedisTemplate<String,String> redisTemplate;
    
    @GetMapping("testLock1")
    public void testLock(){
        // 1 获取锁，setne
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111", 2, TimeUnit.SECONDS);
        // 2 获取锁成功、查询 num 的值
        if(lock){
            Object value = redisTemplate.opsForValue().get("num");
            // 2.1 判断 num 为空 return
            if(StringUtils.isEmpty(value)){
                return;
            }
            // 2.2 有值就转成成 int
            int num = Integer.parseInt(value + "");
            // 2.3 把 redis 的 num 加 1
            redisTemplate.opsForValue().set("num", String.valueOf(++num));
            // 2.4 释放锁，del
            redisTemplate.delete("lock");
        }else{
            // 3 获取锁失败、每隔 0.1 秒再获取
            try {
                Thread.sleep(100);
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

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

出现的问题：如果一个线程（线程 1）获得了锁并执行操作，但是由于某些原因（如网络延迟、GC暂停等）导致操作执行时间超过了锁的过期时间（2秒）。这时，锁会自动过期并释放，使得另一个线程（线程 2）有机会获取到这个锁。如果线程 1 在线程 2 操作完毕前恢复执行并尝试释放锁，它实际上会释放掉线程 2 持有的锁，这就破坏了分布式锁的互斥性，造成资源的不安全访问。

解决方案

在获取锁时，不仅仅设置锁的键，还需要设置一个唯一标识符（如 UUID）作为键的值。这样，当尝试释放锁时，可以先检查键的值是否与当前线程设置的唯一标识符相同，只有在标识符匹配的情况下才执行删除键的操作。

# 优化之UUID防误删

# 代码三（无原子性）

在代码一的基础上，加上了 uuid，看第 23 - 26 行代码

public class RedisLocked {
    @Autowired
    private RedisTemplate<String,String> redisTemplate;
    
    @GetMapping("testLock")
    public void testLocked(){
        String locKey = "lock";
        String uuid = UUID.randomUUID().toString();
        Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 2, TimeUnit.SECONDS);

        if(lock){
            String value = redisTemplate.opsForValue().get("num");
            if(StringUtils.isEmpty(value)){
                return;
            }
            int num = Integer.parseInt(value + "");
            redisTemplate.opsForValue().set("num", String.valueOf(++num));

            // 问题：如果上一行卡顿3秒，而 lock 是 2 秒过期，导致2秒后其他进程拿到锁，而再过 1 秒后这里删除的是其他进程拿的锁
            // redisTemplate.delete(locKey); 

            // 利用UUID判断，解决上面的问题
            if(uuid.equals(redisTemplate.opsForValue().get(locKey))){
                // 新问题：如果进入这一行代码即将执行下面的删除操作，但是 lock 正好过期了，导致下面删除的依然是其他进程拿到的锁
                redisTemplate.delete(locKey);
            }
        }else {
            try {
                Thread.sleep(200);
                testLocked();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

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

可能遇到的问题：

当前线程通过 setIfAbsent 成功设置了带有 UUID 的锁，并正确执行业务逻辑。
在准备释放锁时，通过 UUID 进行了匹配检查，确保是自己的锁。
然而，在执行删除操作的一瞬间，锁可能由于设置的过期时间而自动过期并被删除，此时另一个线程可能刚好获取到了锁。
当前线程执行删除操作时，可能会错误地删除了新线程所持有的锁。

解决方案

使用 Lua 脚本来确保检查 UUID 和删除锁的操作是原子性的。Lua 脚本能够在 Redis 服务器上原子性地执行复杂的操作，这意味着这些操作要么全部执行，要么全部不执行，不存在中间状态。通过将检查和删除锁的逻辑封装到一个 Lua 脚本中并执行该脚本，可以确保：

即使锁自动过期，也不会在检查和删除操作之间插入其他操作，避免误删除其他线程的锁。
只有当 UUID 完全匹配时，才会执行删除操作，确保了锁的安全性和互斥性。

# 代码四（终极版）

public class RedisLocked {
    @Autowired
    private RedisTemplate<String,String> redisTemplate;

    @GetMapping("testLock")
    public void testLocked(){
        String locKey = "lock";
        String uuid = UUID.randomUUID().toString();
        Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 2, TimeUnit.SECONDS);

        if(lock){
            String value = redisTemplate.opsForValue().get("num");
            if(StringUtils.isEmpty(value)){
                return;
            }
            int num = Integer.parseInt(value + "");
            redisTemplate.opsForValue().set("num", String.valueOf(++num));

            // 问题：如果上一行卡顿3秒，而lock 是2秒过期，导致2秒后其他进程拿到锁，而再过1秒后这里删除的是其他进程拿的锁
            // redisTemplate.delete(locKey); 

            // 利用UUID判断，解决上面的问题
            /*if(uuid.equals(redisTemplate.opsForValue().get(locKey))){
                // 新问题：如果进入这一行代码即将执行下面的删除操作，但是lock正好过期了，导致下面删除的依然是其他进程拿到的锁
                redisTemplate.delete(locKey);
            }*/

            /*使用 lua 脚本来解决上面出现的问题*/
            // 定义 lua 脚本
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            // 使用 redis 执行 lua 执行
            DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
            redisScript.setScriptText(script);
            // 设置一下返回值类型 为 Long
            // 因为删除判断的时候，返回的 0,给其封装为数据类型。如果不封装那么默认返回 String 类型，
            // 那么返回字符串与 0 会有发生错误。
            redisScript.setResultType(Long.class);
            // 第一个要是 script 脚本 ，第二个需要判断的 key，第三个就是 key 所对应的值。
            redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid);

        }else {
            try {
                Thread.sleep(200);
                testLocked();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

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

# 4. 总结

Java 代码总结

加锁（setnx 指令）
添加过期时间（setnx 指令加时间）
添加唯一标识如：uuid（将 uuid 放入 Reids，然后操作时获取 uuid，添加 if 判断）
添加原子性，用 LUA 语言实现（第 2、3 步用 LUA 语言编写）

分布式锁总结

为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

互斥性。在任意时刻，只有一个客户端能持有锁
不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁
解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了
加锁和解锁必须具有原子性

编辑此页

上次更新: 2024/12/28, 18:32:08

← Redis - 缓存问题 Redis - Spring Data Redis→