本篇文章小编给大家分享一下redis分布式锁RedissonLock实现代码解析,文章代码介绍的很详细,小编觉得挺不错的,现在分享给大家供大家参考,有需要的小伙伴们可以来看看。
redis分布式锁RedissonLock
简单使用
String key = "key-lock"; RLock lock = redisson.getLock(key); lock.lock(); try { // TODO } catch (Exception e){ log.error(e.getMessage(), e); } finally { lock.unlock(); }
String key = "key-tryLock"; long maxWaitTime = 3_000; RLock lock = redisson.getLock(key); if (lock.tryLock(maxWaitTime, TimeUnit.MILLISECONDS)){ try { // TODO } catch (Exception e){ log.error(e.getMessage(), e); } finally { lock.unlock(); } } else { log.debug("redis锁竞争失败"); }
流程图
多个线程节点锁竞争的正常流程如下图:
多个线程节点锁竞争,并出现节点下线的异常流程如下图:
源码解析
RedissonLock是可重入锁,使用redis的hash结构作为锁的标识存储,锁的名称作为hash的key,UUID + 线程ID作为hash的field,锁被重入的次数作为hash的value。如图所示:
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException { long threadId = Thread.currentThread().getId(); // 尝试获取锁,锁获取成功则ttl为null;获取失败则返回锁的剩余过期时间 Long ttl = tryAcquire(leaseTime, unit, threadId); if (ttl == null) { return; } // 锁被其他线程占用而索取失败,使用线程通知而非自旋的方式等待锁 // 使用redis的发布订阅pub/sub功能来等待锁的释放通知 RFuturefuture = subscribe(threadId); commandExecutor.syncSubscription(future); try { while (true) { ttl = tryAcquire(leaseTime, unit, threadId); // 尝试获取锁,锁获取成功则ttl为null;获取失败则返回锁的剩余过期时间 if (ttl == null) { break; } if (ttl >= 0) { // 使用LockSupport.parkNanos方法线程休眠 try { getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { if (interruptibly) { throw e; } getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } } else { if (interruptibly) { getEntry(threadId).getLatch().acquire(); } else { getEntry(threadId).getLatch().acquireUninterruptibly(); } } } } finally { // 退出锁竞争(锁获取成功或者放弃获取锁),则取消锁的释放订阅 unsubscribe(future, threadId); } }
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException { long time = unit.toMillis(waitTime); long current = System.currentTimeMillis(); long threadId = Thread.currentThread().getId(); Long ttl = tryAcquire(leaseTime, unit, threadId); if (ttl == null) { return true; } time -= System.currentTimeMillis() - current; if (time <= 0) { acquireFailed(threadId); return false; } current = System.currentTimeMillis(); RFuturesubscribeFuture = subscribe(threadId); if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) { if (!subscribeFuture.cancel(false)) { subscribeFuture.onComplete((res, e) -> { if (e == null) { unsubscribe(subscribeFuture, threadId); } }); } acquireFailed(threadId); return false; } try { time -= System.currentTimeMillis() - current; if (time <= 0) { acquireFailed(threadId); return false; } while (true) { long currentTime = System.currentTimeMillis(); ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired if (ttl == null) { return true; } time -= System.currentTimeMillis() - currentTime; if (time <= 0) { acquireFailed(threadId); return false; } currentTime = System.currentTimeMillis(); if (ttl >= 0 && ttl < time) { getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } else { getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS); } time -= System.currentTimeMillis() - currentTime; if (time <= 0) { acquireFailed(threadId); return false; } } } finally { unsubscribe(subscribeFuture, threadId); } }
RedissonLock实现的是可重入锁,通过redis的hash结构实现,而非加单的set nx ex。为了实现原子性的复杂的加锁逻辑,而通过lua脚本实现。获取锁会有如下三种状态:
1、锁未被任何线程占用,则锁获取成功,返回null
2、锁被当前线程占用,则锁获取成功并进行锁的重入,对锁的重入计数+1,返回null
3、锁被其他线程占用,则锁获取失败,返回该锁的自动过期时间ttl
RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) { internalLockLeaseTime = unit.toMillis(leaseTime); return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command, "if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('hset', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "return redis.call('pttl', KEYS[1]);", Collections.
当锁因为被其他线程占用而 使用redis的发布订阅pub/sub功能,通过监听锁的释放通知(在其他线程通过RedissonLock释放锁时,会通过发布订阅pub/sub功能发起通知),等待锁被其他线程释放。通过如此的线程唤醒而非自旋的操作,提高了锁的效率。
public RFuturesubscribe(String entryName, String channelName) { AtomicReference listenerHolder = new AtomicReference (); AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName)); RPromise newPromise = new RedissonPromise () { @Override public boolean cancel(boolean mayInterruptIfRunning) { return semaphore.remove(listenerHolder.get()); } }; Runnable listener = new Runnable() { @Override public void run() { E entry = entries.get(entryName); if (entry != null) { entry.aquire(); semaphore.release(); entry.getPromise().onComplete(new TransferListener (newPromise)); return; } E value = createEntry(newPromise); value.aquire(); E oldValue = entries.putIfAbsent(entryName, value); if (oldValue != null) { oldValue.aquire(); semaphore.release(); oldValue.getPromise().onComplete(new TransferListener (newPromise)); return; } RedisPubSubListener
由于是可重入锁则需要在释放锁的时候做订阅通知,因此释放锁的操作同样是lua脚本实现。锁的释放会有如下三个状态:
1、等待释放的锁不存在或者不是当前线程持有,返回null
2、等待释放的锁被当前线程持有,且该锁当前被重入多次,则锁的重入计数-1,返回0
3、等待释放的锁被当前线程持有,且该锁当前未被重入,则锁的删除并发布该锁释放的订阅通知,返回1
protected RFutureunlockInnerAsync(long threadId) { return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; " + "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + "if (counter > 0) then " + "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " + "else " + "redis.call('del', KEYS[1]); " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; "+ "end; " + "return nil;", Arrays.
Watchdog
RedissonLock为了避免应用获取锁后宕机,因为没人来释放锁而导致死锁情况的出现,默认每次锁的占用只有30秒的时间(org.redisson.config.Config#lockWatchdogTimeout = 30 * 1000)。
于是便有了Watchdog设计,由独立的线程定时给未释放的锁续期,默认锁有效期的三分之一的时长即每10秒给锁自动续期。
private void renewExpiration() { ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (ee == null) { return; } // 默认10秒钟后执行锁续期任务 Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (ent == null) { return; } Long threadId = ent.getFirstThreadId(); if (threadId == null) { return; } RFuturefuture = renewExpirationAsync(threadId); future.onComplete((res, e) -> { if (e != null) { log.error("Can't update lock " + getName() + " expiration", e); return; } // 如果锁续期成功,则10秒钟后再次续期 if (res) { renewExpiration(); } }); } }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); ee.setTimeout(task); } protected RFuture renewExpirationAsync(long threadId) { return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return 1; " + "end; " + "return 0;", Collections.
Redisson 几种锁
1. 可重入锁(Reentrant Lock)
Redisson的分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口,同时还支持自动过期解锁。
public void testReentrantLock(RedissonClient redisson){ RLock lock = redisson.getLock("anyLock"); try{ // 1. 最常见的使用方法 //lock.lock(); // 2. 支持过期解锁功能,10秒钟以后自动解锁, 无需调用unlock方法手动解锁 //lock.lock(10, TimeUnit.SECONDS); // 3. 尝试加锁,最多等待3秒,上锁以后10秒自动解锁 boolean res = lock.tryLock(3, 10, TimeUnit.SECONDS); if(res){ //成功 // do your business } } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } }
Redisson同时还为分布式锁提供了异步执行的相关方法:
public void testAsyncReentrantLock(RedissonClient redisson){ RLock lock = redisson.getLock("anyLock"); try{ lock.lockAsync(); lock.lockAsync(10, TimeUnit.SECONDS); Futureres = lock.tryLockAsync(3, 10, TimeUnit.SECONDS); if(res.get()){ // do your business } } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } finally { lock.unlock(); } }
2. 公平锁(Fair Lock)
Redisson分布式可重入公平锁也是实现了java.util.concurrent.locks.Lock接口的一种RLock对象。在提供了自动过期解锁功能的同时,保证了当多个Redisson客户端线程同时请求加锁时,优先分配给先发出请求的线程。
public void testFairLock(RedissonClient redisson){ RLock fairLock = redisson.getFairLock("anyLock"); try{ // 最常见的使用方法 fairLock.lock(); // 支持过期解锁功能, 10秒钟以后自动解锁,无需调用unlock方法手动解锁 fairLock.lock(10, TimeUnit.SECONDS); // 尝试加锁,最多等待100秒,上锁以后10秒自动解锁 boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } finally { fairLock.unlock(); } }
Redisson同时还为分布式可重入公平锁提供了异步执行的相关方法:
RLock fairLock = redisson.getFairLock("anyLock"); fairLock.lockAsync(); fairLock.lockAsync(10, TimeUnit.SECONDS); Futureres = fairLock.tryLockAsync(100, 10, TimeUnit.SECONDS);
3. 联锁(MultiLock)
Redisson的RedissonMultiLock对象可以将多个RLock对象关联为一个联锁,每个RLock对象实例可以来自于不同的Redisson实例。
public void testMultiLock(RedissonClient redisson1, RedissonClient redisson2, RedissonClient redisson3){ RLock lock1 = redisson1.getLock("lock1"); RLock lock2 = redisson2.getLock("lock2"); RLock lock3 = redisson3.getLock("lock3"); RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3); try { // 同时加锁:lock1 lock2 lock3, 所有的锁都上锁成功才算成功。 lock.lock(); // 尝试加锁,最多等待100秒,上锁以后10秒自动解锁 boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } }
4. 红锁(RedLock)
Redisson的RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个RLock
对象关联为一个红锁,每个RLock对象实例可以来自于不同的Redisson实例。
public void testRedLock(RedissonClient redisson1, RedissonClient redisson2, RedissonClient redisson3){ RLock lock1 = redisson1.getLock("lock1"); RLock lock2 = redisson2.getLock("lock2"); RLock lock3 = redisson3.getLock("lock3"); RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3); try { // 同时加锁:lock1 lock2 lock3, 红锁在大部分节点上加锁成功就算成功。 lock.lock(); // 尝试加锁,最多等待100秒,上锁以后10秒自动解锁 boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } }
5. 读写锁(ReadWriteLock)
Redisson的分布式可重入读写锁RReadWriteLock Java对象实现了java.util.concurrent.locks.ReadWriteLock接口。同时还支持自动过期解锁。该对象允许同时有多个读取锁,但是最多只能有一个写入锁。
RReadWriteLock rwlock = redisson.getLock("anyRWLock"); // 最常见的使用方法 rwlock.readLock().lock(); // 或 rwlock.writeLock().lock(); // 支持过期解锁功能 // 10秒钟以后自动解锁 // 无需调用unlock方法手动解锁 rwlock.readLock().lock(10, TimeUnit.SECONDS); // 或 rwlock.writeLock().lock(10, TimeUnit.SECONDS); // 尝试加锁,最多等待100秒,上锁以后10秒自动解锁 boolean res = rwlock.readLock().tryLock(100, 10, TimeUnit.SECONDS); // 或 boolean res = rwlock.writeLock().tryLock(100, 10, TimeUnit.SECONDS); ... lock.unlock();
6. 信号量(Semaphore)
Redisson的分布式信号量(Semaphore)Java对象RSemaphore采用了与java.util.concurrent.Semaphore相似的接口和用法。
RSemaphore semaphore = redisson.getSemaphore("semaphore"); semaphore.acquire(); //或 semaphore.acquireAsync(); semaphore.acquire(23); semaphore.tryAcquire(); //或 semaphore.tryAcquireAsync(); semaphore.tryAcquire(23, TimeUnit.SECONDS); //或 semaphore.tryAcquireAsync(23, TimeUnit.SECONDS); semaphore.release(10); semaphore.release(); //或 semaphore.releaseAsync();
7. 可过期性信号量(PermitExpirableSemaphore)
Redisson的可过期性信号量(PermitExpirableSemaphore)实在RSemaphore对象的基础上,为每个信号增加了一个过期时间。每个信号可以通过独立的ID来辨识,释放时只能通过提交这个ID才能释放。
RPermitExpirableSemaphore semaphore = redisson.getPermitExpirableSemaphore("mySemaphore"); String permitId = semaphore.acquire(); // 获取一个信号,有效期只有2秒钟。 String permitId = semaphore.acquire(2, TimeUnit.SECONDS); // ... semaphore.release(permitId);
8. 闭锁(CountDownLatch)
Redisson的分布式闭锁(CountDownLatch)Java对象RCountDownLatch采用了与java.util.concurrent.CountDownLatch相似的接口和用法。
忍者必须死34399账号登录版 最新版v1.0.138v2.0.72
下载勇者秘境oppo版 安卓版v1.0.5
下载忍者必须死3一加版 最新版v1.0.138v2.0.72
下载绝世仙王官方正版 最新安卓版v1.0.49
下载Goat Simulator 3手机版 安卓版v1.0.8.2
Goat Simulator 3手机版是一个非常有趣的模拟游
Goat Simulator 3国际服 安卓版v1.0.8.2
Goat Simulator 3国际版是一个非常有趣的山羊模
烟花燃放模拟器中文版 2025最新版v1.0
烟花燃放模拟器是款仿真的烟花绽放模拟器类型单机小游戏,全方位
我的世界动漫世界 手机版v友y整合
我的世界动漫世界模组整合包是一款加入了动漫元素的素材整合包,
我的世界贝爷生存整合包 最新版v隔壁老王
我的世界MITE贝爷生存整合包是一款根据原版MC制作的魔改整