上海著名网站建设,免费ppt背景图,常州到丹阳,什么是网络推广工作基于Redis的分布式MultiLock对象允许对Lock对象进行分组并将它们作为单个锁进行处理。每个RLock对象可能属于不同的Redisson实例。
如果获取的Redisson实例MultiLock崩溃#xff0c;那么它可能永远挂在获取状态。为了避免这种情况#xff0c;Redisson维护了一个锁看门狗那么它可能永远挂在获取状态。为了避免这种情况Redisson维护了一个锁看门狗它会在持有者Redisson实例处于活动状态时延长锁过期时间。默认情况下锁定看门狗超时为30s可以通过Config.lockWatchdogTimeout设置进行更改。作者的另外一篇文章有对看门狗机制有解析基于Redisson的可重入分布式锁
leaseTime在指定的时间间隔后锁将自动释放
MultiLock对象的行为符合java锁规范。这意味着只有锁的拥有者线程才能解锁它否则会抛出IllegalMonitorStateException异常。否则考虑使用RSemaphore对象。
使用示例
普通使用示例
RLock lock1 redisson1.getLock(lock1);
RLock lock2 redisson2.getLock(lock2);
RLock lock3 redisson3.getLock(lock3);RLock multiLock anyRedisson.getMultiLock(lock1, lock2, lock3);// traditional lock method
multiLock.lock();// or acquire lock and automatically unlock it after 10 seconds
multiLock.lock(10, TimeUnit.SECONDS);// or wait for lock aquisition up to 100 seconds
// and automatically unlock it after 10 seconds
boolean res multiLock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
try {
...
} finally {
multiLock.unlock();
}
}Async接口使用的代码示例
RLock lock1 redisson1.getLock(lock1);
RLock lock2 redisson2.getLock(lock2);
RLock lock3 redisson3.getLock(lock3);RLock multiLock anyRedisson.getMultiLock(lock1, lock2, lock3);RFutureVoid lockFuture multiLock.lockAsync();// or acquire lock and automatically unlock it after 10 seconds
RFutureVoid lockFuture multiLock.lockAsync(10, TimeUnit.SECONDS);// or wait for lock aquisition up to 100 seconds
// and automatically unlock it after 10 seconds
RFutureBoolean lockFuture multiLock.tryLockAsync(100, 10, TimeUnit.SECONDS);lockFuture.whenComplete((res, exception) - {// ...multiLock.unlockAsync();
});
Reactive接口使用的代码示例:
RedissonReactiveClient anyRedisson redissonClient.reactive();RLockReactive lock1 redisson1.getLock(lock1);
RLockReactive lock2 redisson2.getLock(lock2);
RLockReactive lock3 redisson3.getLock(lock3);RLockReactive multiLock anyRedisson.getMultiLock(lock1, lock2, lock3);MonoVoid lockMono multiLock.lock();// or acquire lock and automatically unlock it after 10 seconds
MonoVoid lockMono multiLock.lock(10, TimeUnit.SECONDS);// or wait for lock aquisition up to 100 seconds
// and automatically unlock it after 10 seconds
MonoBoolean lockMono multiLock.tryLock(100, 10, TimeUnit.SECONDS);lockMono.doOnNext(res - {// ...
})
.doFinally(multiLock.unlock())
.subscribe();RxJava3接口使用的代码示例:
RedissonRxClient anyRedisson redissonClient.rxJava();RLockRx lock1 redisson1.getLock(lock1);
RLockRx lock2 redisson2.getLock(lock2);
RLockRx lock3 redisson3.getLock(lock3);RLockRx multiLock anyRedisson.getMultiLock(lock1, lock2, lock3);Completable lockRes multiLock.lock();// or acquire lock and automatically unlock it after 10 seconds
Completable lockRes multiLock.lock(10, TimeUnit.SECONDS);// or wait for lock aquisition up to 100 seconds
// and automatically unlock it after 10 seconds
SingleBoolean lockRes multiLock.tryLock(100, 10, TimeUnit.SECONDS);lockRes.doOnSuccess(res - {// ...
})
.doFinally(multiLock.unlock())
.subscribe();源码解析(RedissonMultiLock)
Redisson获取联锁
// 这里相对简单就是创建了一个RLock集合为了后续分别去获取锁
final ListRLock locks new ArrayList();
Override
public RLock getMultiLock(RLock... locks) {return new RedissonMultiLock(locks);
}
public RedissonMultiLock(RLock... locks) {if (locks.length 0) {throw new IllegalArgumentException(Lock objects are not defined);}this.locks.addAll(Arrays.asList(locks));
}加锁
leaseTime指定加锁的时间。超过这个时间后锁便自动解开了。
为了方便我们的源码分析假设我们的locks的size为6。leaseTime为2s
Override
public void lock(long leaseTime, TimeUnit unit) {try {lockInterruptibly(leaseTime, unit);} catch (InterruptedException e) {Thread.currentThread().interrupt();}
}Override
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {// 基础等待时间设置为连锁数量*1500单位是毫秒 // 6*15009000ms 也就是9slong baseWaitTime locks.size() * 1500;// 设置等待时间为-1long waitTime -1;// 如果锁释放的时间为-1就让等待时间等于基础等待时间9s// lock的无参方法默认leaseTime-1if (leaseTime -1) {waitTime baseWaitTime;} else {// 如果锁的释放时间不为-1把leaseTime转为毫秒leaseTime unit.toMillis(leaseTime);// 把锁的释放时间传给等待时间如果leaseTime2s那么waitTime也等于2swaitTime leaseTime;if (waitTime 2000) {// 也就是说leaseTime即使小于2swaitTime也会被重置为2swaitTime 2000;} else if (waitTime baseWaitTime) {// 如果leaseTime大于2s并且小于9s将重新设置等待时间我们暂且还不知道这个等待时间做什么用。// 如果leaseTime等于6那么waitTime6此时waitTime小于9s重新设置waitTime// 将waitTime设置为大于等于3小于6的整数。(此处不明白看下面的解释)waitTime ThreadLocalRandom.current().nextLong(waitTime/2, waitTime);} else {// 如果leaseTime大于2s而且大于9s(baseWaitTime)同样重新设置waitTime的值// 如果传入的leaseTime10s那么waitTime一开始也是10s并且大于baseWaitTime的9s// 将waitTime设置为大于等于9s小于10s的整数。waitTime ThreadLocalRandom.current().nextLong(baseWaitTime, waitTime);}}while (true) {// 传入waitTime开始尝试获取锁了if (tryLock(waitTime, leaseTime, TimeUnit.MILLISECONDS)) {return;}}
}ThreadLocalRandom.current().nextLong(origin, bound)是用于生成一个指定范围内的随机长整数。
具体解释如下
ThreadLocalRandom.current() 返回当前线程的 ThreadLocalRandom 实例用于生成随机数。
nextLong(origin, bound) 生成一个介于 origin包含和 bound不包含之间的随机长整型数。这意味着生成的随机数大于等于 origin并且小于 bound。
此处为什么需要去修改waitTime的值为什么还得整个随机数使用baseWaitTime调整waitTime的作用是什么
尝试获取锁
waitTime表示尝试获取锁的等待时间。它指定了在尝试获取锁时最长的等待时间。
leaseTime 指定加锁的时间。超过这个时间后锁便自动解开了。
TimeUnit时间单位
// 假设传入的waitTime2s leaseTime2s
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {// 定义了一个新的释放时间newLeaseTime-1long newLeaseTime -1;// 如果传入了时间的tryLockleaseTime就不等于-1不传默认值为-1if (leaseTime ! -1) {// 将新的锁释放时间设置为waitTime的2倍单位是毫秒也就是4000msnewLeaseTime unit.toMillis(waitTime)*2;}// 获取当前时间毫秒long time System.currentTimeMillis();// remain保持先翻译为保持时间定义为-1long remainTime -1;if (waitTime ! -1) {// 保持时间设置为waitTime2000msremainTime unit.toMillis(waitTime);}// calcLockWaitTime(remainTime);--return Math.max(remainTime / locks.size(), 1);// 300mslockWaitTimelong lockWaitTime calcLockWaitTime(remainTime);// return 0int failedLocksLimit failedLocksLimit();ListRLock acquiredLocks new ArrayList(locks.size());// 循环获取锁for (ListIteratorRLock iterator locks.listIterator(); iterator.hasNext();) {// 获取到的redisson实例生成的锁RLock lock iterator.next();// 锁获取标识boolean lockAcquired;try {if (waitTime -1 leaseTime -1) {lockAcquired lock.tryLock();} else {// awaitTime300mslong awaitTime Math.min(lockWaitTime, remainTime);// 直接去获取锁返回true or falselockAcquired lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);}} catch (RedisResponseTimeoutException e) {// 如果发生了RedisResponseTimeoutException会先解锁。因为这个时候不确定是否加锁成功了所以解锁设置标识为失败。unlockInner(Arrays.asList(lock));lockAcquired false;} catch (Exception e) {// 其他异常设置标识为falselockAcquired false;}if (lockAcquired) {// 如果加锁成功 放入集合中acquiredLocks.add(lock);} else {// 6-当前成功的数量0直接退出循环也就是说超过了最大的失败限制// 这里RedissonRedLock有重写红锁有自己的规则if (locks.size() - acquiredLocks.size() failedLocksLimit()) {break;}// failedLocksLimit0那么只要失败就进入这个逻辑if (failedLocksLimit 0) {// 会把获取到锁的一次性解锁unlockInner(acquiredLocks);if (waitTime -1 leaseTime -1) {return false;}// 重置failedLocksLimit0failedLocksLimit failedLocksLimit();// 清空获取到锁的集合acquiredLocks.clear();// reset iteratorwhile (iterator.hasPrevious()) {iterator.previous();}} else {// RedissonRedLock才会进入这个逻辑failedLocksLimit--;}}// 如果remainTime不为-1// remainTime2000msif (remainTime ! -1) {// 查看remainTime的剩余时间remainTime - System.currentTimeMillis() - time;// 重置timetime System.currentTimeMillis();// 如果保持时间也就是之前的waitTime小于0也就是说超过了尝试获取锁时最长的等待时间释放所有已获得的锁并返回false加锁失败if (remainTime 0) {unlockInner(acquiredLocks);return false;}}}// 如果没有超过尝试获取锁时最长等待时间并且leaseTime不为-1if (leaseTime ! -1) {// 创建了一个RFuture集合ListRFutureBoolean futures new ArrayList(acquiredLocks.size());for (RLock rLock : acquiredLocks) {//为每个锁设置过期时间是一个异步的操作RFutureBoolean future ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);futures.add(future);}for (RFutureBoolean rFuture : futures) {// 阻塞当前线程同步等待每个异步操作的结果rFuture.syncUninterruptibly();}}return true;}
