redisson分布式锁实现原理_redisson连接池

@Service
public class RedissonManager { 

@Autowired
private RedissonClient redissonClient;
/** * 加锁 * * @param lockKey * @return */
public RLock lock(String lockKey) { 

RLock lock = redissonClient.getLock(lockKey);
lock.lock();
return lock;
}
/** * 释放锁 * * @param lockKey */
public void unlock(String lockKey) { 

RLock lock = redissonClient.getLock(lockKey);
lock.unlock();
}
/** * 释放锁 * * @param lock */
public void unlock(RLock lock) { 

lock.unlock();
}
/** * 带超时的锁 * * @param lockKey * @param timeout 超时时间 单位：秒 */
public RLock lock(String lockKey, int timeout) { 

RLock lock = redissonClient.getLock(lockKey);
lock.lock(timeout, TimeUnit.SECONDS);
return lock;
}
/** * 带超时的锁 * * @param lockKey * @param unit 时间单位 * @param timeout 超时时间 */
public RLock lock(String lockKey, TimeUnit unit, int timeout) { 

RLock lock = redissonClient.getLock(lockKey);
lock.lock(timeout, unit);
return lock;
}
/** * 尝试获取锁 * * @param lockKey * @param waitTime 最多等待时间 * @param leaseTime 上锁后自动释放锁时间 * @return */
public boolean tryLock(String lockKey, int waitTime, int leaseTime) { 

RLock lock = redissonClient.getLock(lockKey);
try { 

return lock.tryLock(waitTime, leaseTime, TimeUnit.SECONDS);
} catch (InterruptedException e) { 

return false;
}
}
/** * 尝试获取锁 * * @param lockKey * @param unit 时间单位 * @param waitTime 最多等待时间 * @param leaseTime 上锁后自动释放锁时间 * @return */
public boolean tryLock(String lockKey, TimeUnit unit, int waitTime, int leaseTime) { 

RLock lock = redissonClient.getLock(lockKey);
try { 

return lock.tryLock(waitTime, leaseTime, unit);
} catch (InterruptedException e) { 

return false;
}
}
}

	private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException { 

// 此处为获取当前线程id
long threadId = Thread.currentThread().getId();
// 核心代码见下图后继续回来走逻辑
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// 此处得到获取锁的结果，正常获取锁则ttl为null，竞争锁时返回锁的过期时间 
if (ttl == null) { 

return;
}
// 此处为订阅锁释放事件，
// 如果当前线程通过 Redis 的 channel 订阅锁的释放事件获取得知已经被释放
// 则会发消息通知待等待的线程进行竞争.
RFuture<RedissonLockEntry> future = subscribe(threadId);
if (interruptibly) { 

commandExecutor.syncSubscriptionInterrupted(future);
} else { 

commandExecutor.syncSubscription(future);
}
try { 

while (true) { 

// 此处循环重试获取锁，直至重新获取锁成功才跳出循环，
// 此种做法阻塞进程，一直处于等待锁手动释放或者超时才继续线程
ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) { 

break;
}
// waiting for message
if (ttl >= 0) { 

try { 

future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) { 

if (interruptibly) { 

throw e;
}
future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
}
} else { 

if (interruptibly) { 

future.getNow().getLatch().acquire();
} else { 

future.getNow().getLatch().acquireUninterruptibly();
}
}
}
} finally { 

// 最后释放订阅事件
unsubscribe(future, threadId);
}
// get(lockAsync(leaseTime, unit));
}

    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { 

return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hincrby', 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.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}

	 @Override
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(waitTime, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) { 

return true;
}
// 以上与lock逻辑一致
// 获取锁失败后，中途tryLock会一直判断中间操作耗时是否已经消耗锁的过期时间，如果消耗完则返回false
time -= System.currentTimeMillis() - current;
if (time <= 0) { 

acquireFailed(waitTime, unit, threadId);
return false;
}
current = System.currentTimeMillis();
// 此处为订阅锁释放事件，
// 如果当前线程通过 Redis 的 channel 订阅锁的释放事件获取得知已经被释放
// 则会发消息通知待等待的线程进行竞争.
RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
// 将订阅阻塞，阻塞时间设置为我们调用tryLock设置的最大等待时间，超过时间则返回false
if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) { 

if (!subscribeFuture.cancel(false)) { 

subscribeFuture.onComplete((res, e) -> { 

if (e == null) { 

unsubscribe(subscribeFuture, threadId);
}
});
}
acquireFailed(waitTime, unit, threadId);
return false;
}
try { 

time -= System.currentTimeMillis() - current;
if (time <= 0) { 

acquireFailed(waitTime, unit, threadId);
return false;
}
// 循环获取锁，但由于上面有最大等待时间限制，基本会在上面返回false
while (true) { 

long currentTime = System.currentTimeMillis();
ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) { 

return true;
}
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) { 

acquireFailed(waitTime, unit, threadId);
return false;
}
// waiting for message
currentTime = System.currentTimeMillis();
if (ttl >= 0 && ttl < time) { 

subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else { 

subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
}
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) { 

acquireFailed(waitTime, unit, threadId);
return false;
}
}
} finally { 

unsubscribe(subscribeFuture, threadId);
}
// return get(tryLockAsync(waitTime, leaseTime, unit));
}