Redis实战篇（四）分布式锁

一、定义

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

• 多进程可见
• 互斥
• 高可用
• 高性能
• 安全性

二、实现

三、基于Redis原理

1、获取锁

• 互斥：确保只能有一个线程获取锁
• 非阻塞：尝试一次，成功返回true，失败返回false

SETNX lock thread1
# 添加锁， NX互斥，EX设置超时时间
SET lock thread1 NX EX 10
1
2
3

2、释放锁

• 手动释放
• 超时释放

DEL lock
1

EXPIRE lock 5
1

四、初级版本代码实现

public interface ILock {

    /**
     * 尝试获取锁
     * @param timeoutSec 锁持有的超时时间，过期后自动释放
     * @return true代表获取锁成功; false代表获取锁失败
     */
    boolean tryLock(long timeoutSec);

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

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

/**
 * 实现分布式锁
 */
public class SimpleRedisLock implements ILock{

    private String name;
    private StringRedisTemplate stringRedisTemplate;


    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("lock:" + name, threadId + "", timeoutSec, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(success);
    }

    /**
     * 释放锁
     */
    @Override
    public void unlock() {
        stringRedisTemplate.delete("lock:" + name);
    }
}
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

		// 创建锁对象
        SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        //尝试获取锁
        boolean isLock = lock.tryLock(1000);
        //判断
        if (!isLock) {
            // 获取锁失败，直接返回失败或者重试
            log.error("不允许重复下单！");
            return;
        }
1
2
3
4
5
6
7
8
9
10

五、改进

1、误删锁

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

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

    @Override
    public boolean tryLock(long timeoutSec) {
        //获取线程标识
        String threadId = ID_PREFIX + Thread.currentThread().getId();
        //获取锁
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent("lock:" + 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("lock:" + name);
        //判断是否与当前线程标识一致
        if (threadId.equals(id)) {
        	//一致，释放锁
            stringRedisTemplate.delete("lock:" + name);
        }
        //不一致，什么都不做
     }
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

2、判断和释放 原子性问题

(1)问题分析

(2)Redis的Lua脚本

Redis提供了Lua脚本在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。

• Redis提供的调用函数

redis.call('命令名称', 'key', '其他参数', ...)
# 示例
redis.call('set', 'name', 'allen')
1
2
3

• 执行脚本

EVAL "脚本" numberkeys  key
# 示例
EVAL "return redis.call('set', 'name', 'allen')" 0
# 脚本中key、value作为参数传递
EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 name allen
1
2
3
4
5

java执行lua脚本

	@Override
	public <T> T execute(RedisScript<T> script, List<K> keys, Object... args) {
		return scriptExecutor.execute(script, keys, args);
	}

1
2
3
4
5

初始化脚本

    private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;

    static {
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        //找到unlock.lua文件
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
        UNLOCK_SCRIPT.setResultType(Long.class);
    }

    @Override
    public void unlock() {
        // 调用lua脚本
        stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX + Thread.currentThread().getId());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

六、进一步优化

可重入锁：同一个线程可以多次获取同一把锁

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

1. 不可重入：同一个线程无法多次获取同一把锁
2. 不可重试：获取锁知常识一次就返回false，没有重试机制
3. 超时释放：锁超时释放，虽然可以避免死锁，但如果是业务执行耗时较长，也会导入锁释放，存在安全隐患
4. 主从一致性：如果redis主从集群存在延迟同步，当主节点宕机时，在从节点中获取不到锁信息

七、Redission

1、入门

（1）引入依赖

        <dependency>
            <groupId>org.redissongroupId>
            <artifactId>redissonartifactId>
            <version>3.13.6version>
        dependency>
1
2
3
4
5

（2）配置客户端

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedisConfig {

    @Bean
    public RedissonClient redissionClient(){
        Config config = new Config();
        config.useSingleServer().setAddress("127.0.0.1:6357").setPassword("123456");
        return Redisson.create(config);
    }
}

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

（3）使用Redission的分布式

    @Autowired
    private RedissonClient redissonClient;

    @Test
    void testString() throws InterruptedException {
        //创建锁对象
        RLock lock = redissonClient.getLock("anyLock");
        //获取锁
        boolean b = lock.tryLock(1, 10, TimeUnit.SECONDS);
        //判断是否成功
        if (b) {
            try {
                System.err.println("执行任务");
            } finally {
                //释放锁
                lock.unlock();
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

2、可重入锁原理

3、不可重试原理

4、释放锁原理

5、总结

Redission分布式原理：

• 可重入：利用hash结构记录线程id和重入次数
• 可重试：利用信号量和PubSub功能实现等待、唤醒、获取锁失效的重试机制
• 超市续约：利用watchDog，每隔一段时间（releaseTime/3）,重置超时时间

6、主从一致性原理

八、总结