Redis分布式锁之红锁的实现

<!-- jdk 1.8+ compatible -->
<dependency>
   <groupId>org.redisson</groupId>
   <artifactId>redisson</artifactId>
   <version>3.9.0</version>
</dependency> 

Config config1 = new Config();
config1.useSingleServer().setAddress("redis://172.0.0.1:5378").setPassword("a123456").setDatabase(0);
RedissonClient redissonClient1 = Redisson.create(config1);

Config config2 = new Config();
config2.useSingleServer().setAddress("redis://172.0.0.1:5379").setPassword("a123456").setDatabase(0);
RedissonClient redissonClient2 = Redisson.create(config2);

Config config3 = new Config();
config3.useSingleServer().setAddress("redis://172.0.0.1:5380").setPassword("a123456").setDatabase(0);
RedissonClient redissonClient3 = Redisson.create(config3);


RLock lock1 = redissonClient1.getLock(lockKey);
RLock lock2 = redissonClient2.getLock(lockKey);
RLock lock3 = redissonClient3.getLock(lockKey);


RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);

try {
  
  boolean res = redLock.tryLock((long)waitTimeout, (long)leaseTime, TimeUnit.SECONDS);
  if (res) {
    //成功获得锁，在这里处理业务
  }
} catch (Exception e) {
  throw new RuntimeException("aquire lock fail");
}finally{
  //无论如何, 最后都要解锁
  redLock.unlock();
}

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
  long newLeaseTime = -1;
  if (leaseTime != -1) {
    newLeaseTime = unit.toMillis(waitTime)*2;
  }
 
  long time = System.currentTimeMillis();
  long remainTime = -1;
  if (waitTime != -1) {
    remainTime = unit.toMillis(waitTime);
  }
  long lockWaitTime = calcLockWaitTime(remainTime);
  
  int failedLocksLimit = failedLocksLimit();
  
  List<RLock> acquiredLocks = new ArrayList<>(locks.size());
  for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
    RLock lock = iterator.next();
    boolean lockAcquired;
    
    try {
      if (waitTime == -1 && leaseTime == -1) {
        lockAcquired = lock.tryLock();
      } else {
        long awaitTime = Math.min(lockWaitTime, remainTime);
        lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
      }
    } catch (RedisResponseTimeoutException e) {
      // 如果抛出这类异常，为了防止加锁成功，但是响应失败，需要解锁所有节点
      unlockInner(Arrays.asList(lock));
      lockAcquired = false;
    } catch (Exception e) {
      // 抛出异常表示获取锁失败
      lockAcquired = false;
    }
   
    if (lockAcquired) {
      
      acquiredLocks.add(lock);
    } else {
      
      if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
        break;
      }

      if (failedLocksLimit == 0) {
        unlockInner(acquiredLocks);
        if (waitTime == -1 && leaseTime == -1) {
          return false;
        }
        failedLocksLimit = failedLocksLimit();
        acquiredLocks.clear();
        // reset iterator
        while (iterator.hASPrevious()) {
          iterator.previous();
        }
      } else {
        failedLocksLimit--;
      }
    }

    
    if (remainTime != -1) {
      remainTime -= System.currentTimeMillis() - time;
      time = System.currentTimeMillis();
      if (remainTime <= 0) {
        unlockInner(acquiredLocks);
        return false;
      }
    }
  }

  if (leaseTime != -1) {
    List<RFuture<Boolean>> futures = new ArrayList<>(acquiredLocks.size());
    for (RLock rLock : acquiredLocks) {
      RFuture<Boolean> future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
      futures.add(future);
    }
   
    for (RFuture<Boolean> rFuture : futures) {
      rFuture.syncUninterruptibly();
    }
  }

  
  return true;
}