RedisLockRegistry分布式锁应用及分析

发表于 | 分类于

1. 关于分布式锁

分布式锁就是应用在分布式环境下多个节点之间进行同步或者协作的锁，分布式锁和普通锁一样，也需要有以下特性：

  • 互斥性，保证只有持有锁的某个线程才能进行操作，即在任意时刻，只有一个节点的客户端能持有分布式锁；
  • 可重入性，在同一个节点进程内，同一个线程可多次获取锁；
  • 超时处理机制，需要支持超时自动释放锁，避免死锁的产生，以及避免其他节点长期等待造成的资源浪费；
  • 锁释放机制，加锁和解锁必须是节点内的同一个线程；

2. RedisLockRegistry上手

RedisLockRegistry是Spring-Integration集成工具包项目提供的基于Redis的分布式锁管理器，使用时，首先导入依赖：

1
2
3
4
5
6
7
8
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-redis</artifactId>
</dependency>

其次配置分布式锁：

1
2
3
4
5
6
7
8
9
10
11
/**
 * 配置分布式锁
 *
 * @param redisConnectionFactory
 * @return
 */
@Bean(destroyMethod = "destroy")
public RedisLockRegistry redisLockRegistry(RedisConnectionFactory redisConnectionFactory) {
    long defaultExpireTime = 10000L;
    return new RedisLockRegistry(redisConnectionFactory, "redis-lock", defaultExpireTime);
}

RedisLockRegistry相当于一个锁的管理器，所有的分布式锁都可以从中获取，如上定义，锁的键名为“redis-lock:你定义的key”，超时时间也可以自己设定，默认超时时间是60s。

使用分布式加锁时，只需要参考如下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 锁测试
public void test(String lockKey) {
    // 获取锁
    Lock lock = redisLockRegistry.obtain(lockKey);
    // 加锁
    lock.lock();
    try {
        // 此处是你的代码逻辑，处理需要加锁的一些事务
    } catch (Exception e) {
    } finally {
        // 配合解锁逻辑
        lock.unlock();
    }
}

RedisLockRegistry是基于Redis的setnx和ReentrantLock可重入锁实现。下一章节，我们可以对其源码展开阅读及分析。

3. RedisLockRegistry源码分析

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

/**
 * Constructs a lock registry with the default (60 second) lock expiration.
 * @param connectionFactory The connection factory.
 * @param registryKey The key prefix for locks.
 */
public RedisLockRegistry(RedisConnectionFactory connectionFactory, String registryKey) {
	this(connectionFactory, registryKey, DEFAULT_EXPIRE_AFTER);
}

/**
 * Constructs a lock registry with the supplied lock expiration.
 * @param connectionFactory The connection factory.
 * @param registryKey The key prefix for locks.
 * @param expireAfter The expiration in milliseconds.
 */
public RedisLockRegistry(RedisConnectionFactory connectionFactory, String registryKey, long expireAfter) {
	Assert.notNull(connectionFactory, "'connectionFactory' cannot be null");
	Assert.notNull(registryKey, "'registryKey' cannot be null");
	this.redisTemplate = new StringRedisTemplate(connectionFactory);
	this.obtainLockScript = new DefaultRedisScript<>(OBTAIN_LOCK_SCRIPT, Boolean.class);
	this.registryKey = registryKey;
	this.expireAfter = expireAfter;
}

其中OBTAIN_LOCK_SCRIPT是一个上锁的lua脚本，因为若你在应用层面是分步骤的get/set/expire操作，是不符合原子性的，如果SETNX成功，在服务器挂掉、重启或网络问题等，导致EXPIRE命令没有执行，锁没有设置超时时间，后续就有可能变成死锁，所以最好的方式是通过lua脚本来实现加锁的操作。其lua加锁脚本为：

image

其中KEYS[1]代表当前锁的key值，ARGV[1]代表当前的客户端标识，ARGV[2]代表过期时间。首先根据KEYS[1]从redis中拿到对应的客户端标识，如果已存在的客户端标识和ARGV[1]相等，那么重置过期时间为ARGV[2]；如果值不存在，设置KEYS[1]对应的值为ARGV[1]，并且过期时间设置ARGV[2]，从逻辑上来说，这就是一个简单的get和setnx操作。

3.2 获取锁

获取锁的代码如下：

1
2
3
4
5
6
7
8
private final Map<String, RedisLock> locks = new ConcurrentHashMap<>();

@Override
public Lock obtain(Object lockKey) {
	Assert.isInstanceOf(String.class, lockKey);
	String path = (String) lockKey;
	return this.locks.computeIfAbsent(path, RedisLock::new);
}

RedisLockRegistry维护了一个key-RedisLock类型的ConcurrentHashMap，即在RedisLockRegistry中，每个key对应一个RedisLock。

3.3 RedisLock

RedisLock是RedisLockRegistry的内部实现类，实现了Lock接口，是锁的定义和实现逻辑落地的类。首先看加锁过程：

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
private final ReentrantLock localLock = new ReentrantLock();

@Override
public void lock() {
    // 尝试获取可重入锁localLock
	this.localLock.lock();
	while (true) {
		try {
		    // 尝试获取分布式锁
			while (!obtainLock()) {
			    // 每隔100ms便会重新发起请求分布式锁
				Thread.sleep(100);
			}
			break;
		}
		catch (InterruptedException e) {
			/*
			 * This method must be uninterruptible so catch and ignore
			 * interrupts and only break out of the while loop when
			 * we get the lock.
			 */
		}
		catch (Exception e) {
			this.localLock.unlock();
			rethrowAsLockException(e);
		}
	}
}

从代码可以看到，lock方法首先尝试获取ReentrantLock，如果获取，再尝试去获取分布式锁，使用localLock的目的在于减少节点本地多线程竞争分布式锁，使得每刻只有一个线程去竞争分布式锁，以减少不必要的资源开销，减轻Redis的压力。

本地线程如果获取不到分布式锁，则进行阻塞，直至获取到锁或者出现异常，所以每隔100毫秒会去尝试获取分布式锁，直到获取成功或者抛出异常为止。我们再来看下obtainLock方法的内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
private final long expireAfter;

private final String clientId = UUID.randomUUID().toString();

private boolean obtainLock() {
    
    // 执行加锁的lua脚本
	Boolean success =
			RedisLockRegistry.this.redisTemplate.execute(RedisLockRegistry.this.obtainLockScript,
					Collections.singletonList(this.lockKey), RedisLockRegistry.this.clientId,
					String.valueOf(RedisLockRegistry.this.expireAfter));

	boolean result = Boolean.TRUE.equals(success);

	if (result) {
	    // 记录加锁时间
		this.lockedAt = System.currentTimeMillis();
	}
	return result;
}

获取锁的过程比较简单，通过redisTemplate执行lua脚本获取Redis锁。

RedisLock也定义了可中断锁的过程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Override
public void lockInterruptibly() throws InterruptedException {
	this.localLock.lockInterruptibly();
	try {
		while (!obtainLock()) {
			Thread.sleep(100); //NOSONAR
		}
	}
	catch (InterruptedException ie) {
		this.localLock.unlock();
		Thread.currentThread().interrupt();
		throw ie;
	}
	catch (Exception e) {
		this.localLock.unlock();
		rethrowAsLockException(e);
	}
}

lock方法不会响应中断信号，lockInterruptibly方法利用ReentrantLock的可中断机制，会响应中断信号，即假如获取锁的过程如果出现中断，则结束获取操作过程。

3.4 解锁过程

解锁的方法代码如下：

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
// 解锁的入口
@Override
public void unlock() {

    // 判断localLock锁是否被当前线程持有
	if (!this.localLock.isHeldByCurrentThread()) {
		throw new IllegalStateException("You do not own lock at " + this.lockKey);
	}
	
	// 当前线程持有的可重入锁的数量，即重入的次数
	// 如果此时 > 1，表示当前线程有多次获取可重入锁，释放的时候只减少本地锁的次数，不能释放分布式锁
	if (this.localLock.getHoldCount() > 1) {
		this.localLock.unlock();
		return;
	}
	try {
	    // 判断分布式锁的所有者是否是当前RedisLockRegistry实例
		if (!isAcquiredInThisProcess()) {
			throw new IllegalStateException("Lock was released in the store due to expiration. " +
					"The integrity of data protected by this lock may have been compromised.");
		}
        
        // 响应中断机制
		if (Thread.currentThread().isInterrupted()) {
			RedisLockRegistry.this.executor.execute(this::removeLockKey);
		}
		else {
		    // 直接删除key，释放锁
			removeLockKey();
		}

		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Released lock; " + this);
		}
	}
	catch (Exception e) {
		ReflectionUtils.rethrowRuntimeException(e);
	}
	finally {
	    // 释放本地可重入锁
		this.localLock.unlock();
	}
}

// 判断分布式锁的所有者是否是当前RedisLockRegistry实例
public boolean isAcquiredInThisProcess() {
	return RedisLockRegistry.this.clientId.equals(
			RedisLockRegistry.this.redisTemplate.boundValueOps(this.lockKey).get());
}

// 删除key
private void removeLockKey() {
    // 检查对应的Redis是否支持UNLINK命令
    // 该命令用于异步删除某个键，功能等同于del命
    // 非阻塞，只有在Redis4及以上版本才支持
	if (this.unlinkAvailable) {
		try {
			RedisLockRegistry.this.redisTemplate.unlink(this.lockKey);
		}
		catch (Exception ex) {
			LOGGER.warn("The UNLINK command has failed (not supported on the Redis server?); " +
					"falling back to the regular DELETE command", ex);
			this.unlinkAvailable = false;
			RedisLockRegistry.this.redisTemplate.delete(this.lockKey);
		}
	}
	else {
		RedisLockRegistry.this.redisTemplate.delete(this.lockKey);
	}
}

4. 使用总结

本文主要介绍了基于的分布式锁实现过程及简单源码阅读，其实基于Redis的分布式锁实现，主要是依托get和setnx的方法，再包裹一层本地的可重入锁即可。若有问题，也欢迎各位积极交流。

5. 参考文档

以上内容就是关于RedisLockRegistry分布式锁应用及分析的全部内容了，谢谢你阅读到了这里！

Author：zhaoyh