RedisLockRegistry源码redis分布式锁

redis实现分布式锁&＃xff0c;实现了Lock接口&＃xff0c;和ReentrantLock意向&＃xff0c;有可重入&＃xff0c;阻塞等功能

使用

依赖

org.springframework.bootspring-boot-starter-data-redisredis.clientsjedisio.lettucelettuce-coreredis.clientsjedisorg.springframework.integrationspring-integration-redis


配置类

&＃64;Configuration
public class RedisConfig {&＃64;Beanpublic RedisTemplate redisTemplate(JedisConnectionFactory jedisConnectionFactory ) {//设置序列化Jackson2JsonRedisSerializer jackson2JsonRedisSerializer &＃61; new Jackson2JsonRedisSerializer(Object.class);ObjectMapper om &＃61; new ObjectMapper();om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);jackson2JsonRedisSerializer.setObjectMapper(om);// 配置redisTemplateRedisTemplate redisTemplate &＃61; new RedisTemplate<>();redisTemplate.setConnectionFactory(jedisConnectionFactory);RedisSerializer stringSerializer &＃61; new StringRedisSerializer();redisTemplate.setKeySerializer(stringSerializer); // key序列化redisTemplate.setValueSerializer(stringSerializer); // value序列化redisTemplate.setHashKeySerializer(stringSerializer); // Hash key序列化redisTemplate.setHashValueSerializer(stringSerializer); // Hash value序列化redisTemplate.afterPropertiesSet();return redisTemplate;}&＃64;Beanpublic RedisLockRegistry redisLockRegistry(JedisConnectionFactory jedisConnectionFactory) {return new RedisLockRegistry(jedisConnectionFactory, "REDIS_LOCK");}}

举例

&＃64;RestController
public class RedisController {&＃64;Autowiredprivate RedisLockRegistry redisLockRegistry;&＃64;Autowiredprivate UserService userService;//http://localhost:9000/redisLock?id&＃61;1&＃64;RequestMapping("/redisLock")public String redisLock(Integer id){//redis的key冒号&＃xff1a;连接//registryKey和lockKey自动冒号连接&＃xff0c;最终key为REDIS_LOCK:USER_ID:1&＃xff0c;值为uuidLock lock &＃61; redisLockRegistry.obtain("USER_ID:" &＃43; id);for (int i &＃61; 0; i <3; i&＃43;&＃43;) {new Thread(() -> {lock.lock();System.out.println(Thread.currentThread().getName() &＃43; " begin " &＃43; new Date());userService.update();System.out.println(Thread.currentThread().getName() &＃43; " end " &＃43; new Date());lock.unlock();}).start();}return "ok";}}

Thread-14 begin Fri Jul 19 17:04:30 CST 2019
Thread-14 end Fri Jul 19 17:04:31 CST 2019
Thread-15 begin Fri Jul 19 17:04:31 CST 2019
Thread-15 end Fri Jul 19 17:04:32 CST 2019
Thread-16 begin Fri Jul 19 17:04:32 CST 2019
Thread-16 end Fri Jul 19 17:04:33 CST 2019

源码分析

ExpirableLockRegistry接口&＃xff0c;添加一个过期释放锁的方法

public interface ExpirableLockRegistry extends LockRegistry {/*** Remove locks last acquired more than &＃39;age&＃39; ago that are not currently locked.* &＃64;param age the time since the lock was last obtained.* &＃64;throws IllegalStateException if the registry configuration does not support this feature.*/void expireUnusedOlderThan(long age);}

LockRegistry接口&＃xff0c;只有一个获取锁的方法

&＃64;FunctionalInterface
public interface LockRegistry {/*** Obtains the lock associated with the parameter object.* &＃64;param lockKey The object with which the lock is associated.* &＃64;return The associated lock.*/Lock obtain(Object lockKey);}


RedisLockRegistry构造器

private static final long DEFAULT_EXPIRE_AFTER &＃61; 60000L;private final String registryKey;private final StringRedisTemplate redisTemplate;private final RedisScript obtainLockScript;private final long expireAfter;private static final String OBTAIN_LOCK_SCRIPT &＃61;"local lockClientId &＃61; redis.call(&＃39;GET&＃39;, KEYS[1])\n" &＃43;"if lockClientId &＃61;&＃61; ARGV[1] then\n" &＃43;" redis.call(&＃39;PEXPIRE&＃39;, KEYS[1], ARGV[2])\n" &＃43;" return true\n" &＃43;"elseif not lockClientId then\n" &＃43;" redis.call(&＃39;SET&＃39;, KEYS[1], ARGV[1], &＃39;PX&＃39;, ARGV[2])\n" &＃43;" return true\n" &＃43;"end\n" &＃43;"return false";/*** Constructs a lock registry with the default (60 second) lock expiration.* &＃64;param connectionFactory The connection factory.* &＃64;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.* &＃64;param connectionFactory The connection factory.* &＃64;param registryKey The key prefix for locks.* &＃64;param expireAfter The expiration in milliseconds.*/public RedisLockRegistry(RedisConnectionFactory connectionFactory, String registryKey, long expireAfter) {Assert.notNull(connectionFactory, "&＃39;connectionFactory&＃39; cannot be null");Assert.notNull(registryKey, "&＃39;registryKey&＃39; cannot be null");this.redisTemplate &＃61; new StringRedisTemplate(connectionFactory);this.obtainLockScript &＃61; new DefaultRedisScript<>(OBTAIN_LOCK_SCRIPT, Boolean.class);this.registryKey &＃61; registryKey;this.expireAfter &＃61; expireAfter;}


获取锁

private final Map locks &＃61; new ConcurrentHashMap<>();&＃64;Overridepublic Lock obtain(Object lockKey) {Assert.isInstanceOf(String.class, lockKey);String path &＃61; (String) lockKey;return this.locks.computeIfAbsent(path, RedisLock::new);}


Map

default V computeIfAbsent(K key,Function mappingFunction) {Objects.requireNonNull(mappingFunction);V v;if ((v &＃61; get(key)) &＃61;&＃61; null) {V newValue;if ((newValue &＃61; mappingFunction.apply(key)) !&＃61; null) {put(key, newValue);return newValue;}}return v;}default V putIfAbsent(K key, V value) {V v &＃61; get(key);if (v &＃61;&＃61; null) {v &＃61; put(key, value);}return v;}


每个lockKey创建一个锁&＃xff0c;缓存起来
computeIfAbsent和putIfAbsent的区别是&＃xff0c;前者是一个函数式接口&＃xff0c;创建对象&＃xff0c;作为缓存的值&＃xff0c;后者是直接传进来值

上锁

&＃64;Overridepublic void lock() {this.localLock.lock();while (true) {try {while (!obtainLock()) {Thread.sleep(100); //NOSONAR}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);}}}private final String clientId &＃61; UUID.randomUUID().toString();private boolean obtainLock() {boolean success &＃61; RedisLockRegistry.this.redisTemplate.execute(RedisLockRegistry.this.obtainLockScript,Collections.singletonList(this.lockKey), RedisLockRegistry.this.clientId,String.valueOf(RedisLockRegistry.this.expireAfter));if (success) {this.lockedAt &＃61; System.currentTimeMillis();}return success;}


先用ReentrantLock加锁&＃xff0c;再用redis调用lua脚本&＃xff0c;

private static final String OBTAIN_LOCK_SCRIPT &＃61;"local lockClientId &＃61; redis.call(&＃39;GET&＃39;, KEYS[1])\n" &＃43;"if lockClientId &＃61;&＃61; ARGV[1] then\n" &＃43;" redis.call(&＃39;PEXPIRE&＃39;, KEYS[1], ARGV[2])\n" &＃43;" return true\n" &＃43;"elseif not lockClientId then\n" &＃43;" redis.call(&＃39;SET&＃39;, KEYS[1], ARGV[1], &＃39;PX&＃39;, ARGV[2])\n" &＃43;" return true\n" &＃43;"end\n" &＃43;"return false";


如果lockKey没有值&＃xff0c;设置值&＃xff0c;过期时间60秒。否则是线程重入锁&＃xff0c;刷新过期时间60秒
redis加锁成功后&＃xff0c;每个线程保存加锁时间

如果加锁失败&＃xff0c;100毫秒重试&＃xff0c;一直循环到获取锁&＃xff0c;所以锁是可重入的。

释放锁

RedisLockRegistry.RedisLock

&＃64;Overridepublic void unlock() {if (!this.localLock.isHeldByCurrentThread()) {throw new IllegalStateException("You do not own lock at " &＃43; this.lockKey);}if (this.localLock.getHoldCount() > 1) {this.localLock.unlock();return;}try {if (Thread.currentThread().isInterrupted()) {RedisLockRegistry.this.executor.execute(this::removeLockKey);}else {removeLockKey();}if (logger.isDebugEnabled()) {logger.debug("Released lock; " &＃43; this);}}catch (Exception e) {ReflectionUtils.rethrowRuntimeException(e);}finally {this.localLock.unlock();}}private void removeLockKey() {if (RedisUtils.isUnlinkAvailable(RedisLockRegistry.this.redisTemplate)) {RedisLockRegistry.this.redisTemplate.unlink(this.lockKey);}else {RedisLockRegistry.this.redisTemplate.delete(this.lockKey);}}


ReentrantLock保存了上锁的线程&＃xff0c;和线程的重入次数
如果是重入锁&＃xff0c;计数器减一&＃xff0c;即aqs的state减一
否则redis删除key&＃xff0c;然后释放ReentrantLock锁。