vlambda博客
学习文章列表

 Jedis jedis = jedisPool.getResource(); // 如果锁不存在则进行加锁 Long lockResult = jedis.setnx(lockName, myRandomValue); if (lockResult == 1) { // 设置锁过期时间，加锁和设置过期时间是两步完成的，非原子操作 jedis.expire(lockName, expireTime); }

String script = "if redis.call('get', KEYS[1]) == ARGV[1]  then return redis.call('del', KEYS[1])  else return 0  end";

 Long unlock = 1L; Jedis jedis = null; // Lua脚本，用于校验并释放锁 String script = "if redis.call('get', KEYS[1]) == ARGV[1]  then return redis.call('del', KEYS[1])  else  return 0 end"; try { jedis = jedisPool.getResource(); // 通过 Redis 的 eval() 函数执行 Lua 脚本， // 入参 lockName 赋值给参数 KEYS[1]，myRandomValue 赋值给 ARGV[1]， // eval() 函数将 Lua 脚本交给 Redis 服务端执行。 Object result =  jedis.eval(script,  Collections.singletonList(lockName), Collections.singletonList(myRandomValue));
 // 注意:如果脚本顺利执行将返回1， // 如果执行脚本时，其它的客户端对这个lockName对应的值进行了更改 // 则返回0 if (unlock.equals(result) { return true; } } catch (Exception e) { throw e; } finally { if (null != jedis) { jedis.close(); } } return false;

 Jedis jedis = null;  try { jedis = jedisPool.getResource(); // 监控锁对应的Key，如果其它的客户端对这个Key进行了更改，那么本次事务会被取消。 jedis.watch(lockName); // 成功获取锁，则操作公共资源执行自定义流程 // ...自定义流程代码省略...
 // 校验是否持有锁 if (lockValue.equals(jedis.get(lockName))) { // 开启事务功能， Transaction multi = jedis.multi(); // 释放锁 multi.del(lockName); // 执行事务（如果其它的客户端对这个Key进行了更改，那么本次事务会被取消,不会执行) // 如果正常执行，由于只有一个删除操作，返回的list将只有一个对象。 List<Object> result = multi.exec(); if (RELEASE_SUCCESS.equals(result.size())) { return true; } } } catch (Exception e) { throw e; } finally { if (null != jedis) { jedis.unwatch(); jedis.close(); } }

 Jedis jedis = jedisPool.getResource(); jedis.del(lockName);

Jedis jedis = jedisPool.getResource();if (lockValue.equals(jedis.get(lockName))) { jedis.del(lockName);}

RLock lock = redisson.getLock("anyLock");// 最常见的使用方法lock.lock();

// 显式制定解锁时间，无需调用unlock方法手动解锁lock.lock(10, TimeUnit.SECONDS);// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);if (res) {try { ...} finally { lock.unlock();}

public interface RLock extends Lock, RLockAsync { ......

RLock fairLock = redisson.getFairLock("anyLock");// 最常见的使用方法fairLock.lock();

// 10秒钟以后自动解锁// 无需调用unlock方法手动解锁fairLock.lock(10, TimeUnit.SECONDS);
// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS);...fairLock.unlock();

RLock lock1 = redissonInstance1.getLock("lock1");RLock lock2 = redissonInstance2.getLock("lock2");RLock lock3 = redissonInstance3.getLock("lock3");
RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);// 同时加锁：lock1 lock2 lock3// 所有的锁都上锁成功才算成功。lock.lock();...lock.unlock();

RLock lock1 = redissonInstance1.getLock("lock1");RLock lock2 = redissonInstance2.getLock("lock2");RLock lock3 = redissonInstance3.getLock("lock3");
RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);// 同时加锁：lock1 lock2 lock3// 红锁在大部分节点上加锁成功就算成功。lock.lock();...lock.unlock();

RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);// 给lock1，lock2，lock3加锁，如果没有手动解开的话，10秒钟后将会自动解开lock.lock(10, TimeUnit.SECONDS);
// 为加锁等待100秒时间，并在加锁成功10秒钟后自动解开boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);...lock.unlock();

RReadWriteLock rwlock = redisson.getReadWriteLock("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();

redisson-master |-redisson |-src |-main |-java |-org.redisson

 RLock lock = redisson.getLock(lockName); boolean getLock = false; try { getLock = rLock.tryLock(0, expireSeconds, TimeUnit.SECONDS); if (getLock) { LOGGER.info("获取Redisson分布式锁[成功],lockName={}", lockName); } else { LOGGER.info("获取Redisson分布式锁[失败],lockName={}", lockName); } } catch (InterruptedException e) { LOGGER.error("获取Redisson分布式锁[异常]，lockName=" + lockName, e); e.printStackTrace(); return false; } return getLock;

@Overridepublic RLock getLock(String name) { return new RedissonLock(connectionManager.getCommandExecutor(), name, id);}

protected final UUID id = UUID.randomUUID();

public class RedissonLock extends RedissonExpirable implements RLock {
 ...省略部分代码...
 protected static final LockPubSub PUBSUB = new LockPubSub();
 final CommandAsyncExecutor commandExecutor;
 public RedissonLock(CommandAsyncExecutor commandExecutor, String name, UUID id) { super(commandExecutor, name); this.commandExecutor = commandExecutor; this.id = id; // 看门狗锁续约检查时间周期，默认30s this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(); }

 boolean getLock = rLock.tryLock(0, expireSeconds, TimeUnit.SECONDS);

@Overridepublic boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException { long time = unit.toMillis(waitTime); long current = System.currentTimeMillis(); final long threadId = Thread.currentThread().getId();
 // 申请锁，返回还剩余的锁过期时间 Long ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired // 如果ttl为空，表示锁申请成功 if (ttl == null) { return true; }
 time -= (System.currentTimeMillis() - current); if (time <= 0) { acquireFailed(threadId); return false; }
 current = System.currentTimeMillis();
 // 订阅监听redis的消息，并创建RedissonLockEntry // 其中，RedissonLockEntry中比较关键的是一个Semaphore // 属性对象用来控制本地的锁的请求的信号量同步，返回Netty框架的Future final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
 // 阻塞等待subscribe的future的结果对象，如果subscribe方法调用超过了time， // 说明已经超过了客户端设置的最大的wait time，直接返回false，取消订阅，并且不会再继续申请锁 if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) { if (!subscribeFuture.cancel(false)) { subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() { @Override public void operationComplete(Future<RedissonLockEntry> future) throws Exception { if (subscribeFuture.isSuccess()) { 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; }
 // waiting for message // 阻塞等待锁 currentTime = System.currentTimeMillis();
 // 通过信号量（共享锁）进行阻塞，等待解锁消息 // 如果剩余时间 TTL 小于wait time，就在ttl时间内 // 从Entry的信号量获取一个许可（除非发生中断或者一直不存在可用的许可） // 否则就在wait time时间范围内等待可以通过的信号量 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) { // 等待时间小于等于0，不等待申请锁并返回 acquireFailed(threadId); return false; } } } finally { // 无论最终获取锁是否成功，都需要取消订阅解锁消息，防止死锁发生。 unsubscribe(subscribeFuture, threadId); }}

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) { return get(tryAcquireAsync(leaseTime, unit, threadId));}

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) { if (leaseTime != -1) { return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); } ...省略部分逻辑...}

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { internalLockLeaseTime = unit.toMillis(leaseTime);
 return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command, // 检查key是否已被占用，如果没有则设置超时时间及唯一标识，初始化value=1 "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; " + // 锁重入的情况，判断锁的key field，一致的话，value加1 "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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));}

@Overridepublic void unlock() { // 解锁核心逻辑 Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId())); // 解锁返回空，抛出异常 if (opStatus == null) { throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + id + " thread-id: " + Thread.currentThread().getId()); } if (opStatus) { // 解锁成功之后取消更新锁expire的时间的任务 cancelExpirationRenewal(); }}

void cancelExpirationRenewal() { Timeout task = expirationRenewalMap.remove(getEntryName()); if (task != null) { task.cancel(); }}

protected RFuture<Boolean> unlockInnerAsync(long threadId) { return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, // 如果锁的key已经不存在，表明锁已经被解锁，直接发布redis消息 "if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; " + "end;" +
 // key和field不匹配，说明当前的客户端线程并没有持有锁，不能进行主动解锁操作。 "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; " +
 // 将value减1 "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
 // 如果counter>0表明锁进行了重入，不能删除key，也就是不进行解锁操作 "if (counter > 0) then " + "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " +
 // 否则删除key并发布解锁消息进行解锁 "else " + "redis.call('del', KEYS[1]); " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; "+ "end; " + "return nil;", Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));
}

public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {
 // 解锁消息 public static final Long UNLOCK_MESSAGE = 0L; public static final Long READ_UNLOCK_MESSAGE = 1L;
 ...省略部分逻辑...
 @Override protected void onMessage(RedissonLockEntry value, Long message) { // 如果订阅的消息为解锁消息，UNLOCK_MESSAGE = 0L if (message.equals(UNLOCK_MESSAGE)) { Runnable runnableToExecute = value.getListeners().poll(); if (runnableToExecute != null) { runnableToExecute.run(); } // 释放一个许可，并唤醒等待entry. value.getLatch().release(); }  ...... }
}

@Overridepublic void lock() { try { lockInterruptibly(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }}

@Overridepublic void lockInterruptibly() throws InterruptedException { lockInterruptibly(-1, null);}

@Overridepublic void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException { long threadId = Thread.currentThread().getId();
 // 尝试获取锁 Long ttl = tryAcquire(leaseTime, unit, threadId);
 // 锁获取成功 if (ttl == null) { return; }
 // 通过异步方式订阅Redis的channel，阻塞方式获取订阅结果 RFuture<RedissonLockEntry> future = subscribe(threadId); commandExecutor.syncSubscription(future);
 try { // 通过循环判断，直到锁获取成功，经典写法。 while (true) { ttl = tryAcquire(leaseTime, unit, threadId); // 锁获取成功，跳出循环 if (ttl == null) { break; }
 // 如果剩余时间 TTL 大于0，从Entry的信号量获取一个许可（除非发生中断或者一直不存在可用的许可） // 否则就在wait time时间范围内等待可以通过的信号量 if (ttl >= 0) { getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } else { getEntry(threadId).getLatch().acquire(); } } } finally { // 无论最终获取锁是否成功，都需要取消订阅解锁消息，防止死锁发生。 unsubscribe(future, threadId); }}