redisson 분포식 자물쇠 원리 실현

Redisson 분산 잠금
이전의 주석 기반 자물쇠는 기본적인 Redis의 분포식 자물쇠였다. 자물쇠의 실현은 Redisson 구성 요소가 제공하는 RLock을 바탕으로 했다. 이 부분에서 Redisson이 자물쇠를 어떻게 실현하는지 살펴보자.
서로 다른 버전이 자물쇠를 실현하는 메커니즘은 결코 같지 않다
인용된 Redisson이 최근에 발표한 버전 3.2.3은 서로 다른 버전이 자물쇠를 실현할 수 있는 메커니즘이 같지 않다. 초기 버전은 간단한 setnx, getset 등 일반적인 명령으로 설정된 것 같았고 후기에 Redis가 스크립트 Lua를 지원하여 실현 원리를 변경했다.

<dependency>
 <groupId>org.redisson</groupId>
 <artifactId>redisson</artifactId>
 <version>3.2.3</version>
</dependency>

setnx는 getset과 업무를 협조하여 완성해야 자물쇠 문제를 비교적 잘 피할 수 있습니다. 새로운 버전은 루아 스크립트를 지원하기 때문에 업무를 피하고 여러 개의 redis 명령을 조작할 수 있으며 의미 표현이 더욱 명확합니다.
RLOCK 인터페이스의 특징
상속 표준 인터페이스 Lock
표준 자물쇠 인터페이스의 모든 기능을 가지고 있습니다. 예를 들어lock,unlock,trylock 등입니다.
확장 표준 인터페이스 Lock
많은 방법을 확장했는데 주로 강제 자물쇠 방출, 유효기간이 있는 자물쇠, 그리고 비동기적인 방법이 있다.그 중에서 앞의 두 가지 방법은 주로 표준lock이 초래할 수 있는 자물쇠 문제를 해결하는 것이다.예를 들어 어떤 라인이 자물쇠를 얻은 후에 라인이 있는 기계가 다운되었다. 이때 자물쇠를 얻은 라인이 자물쇠를 정상적으로 방출하지 못해 나머지 자물쇠를 기다리는 라인이 계속 기다린다.
재입력 가능 메커니즘
각 버전의 구현은 차이가 있다. 재입력은 주로 성능을 고려한다. 같은 라인이 자물쇠를 풀지 않았을 때 다시 자물쇠 자원을 신청하면 신청 절차를 밟지 않고 이미 가져온 자물쇠를 계속 되돌려주고 재입력한 횟수를 기록하면 된다. jdk의 ReentrantLock 기능과 유사하다.리셋 횟수는 hincrby 명령에 따라 사용됩니다. 상세한 매개 변수 아래의 코드입니다.
어떻게 같은 노선이라고 판단합니까?
redisson의 방안은 RedissonLock의 실례적인 guid에 현재 라인의 id를 추가하여 getLockName을 통해 되돌아오는 것입니다.

public class RedissonLock extends RedissonExpirable implements RLock {
 final UUID id;
 protected RedissonLock(CommandExecutor commandExecutor, String name, UUID id) {
  super(commandExecutor, name);
  this.internalLockLeaseTime = TimeUnit.SECONDS.toMillis(30L);
  this.commandExecutor = commandExecutor;
  this.id = id;
 }
 String getLockName(long threadId) {
  return this.id + ":" + threadId;
 }

RLOCK 자물쇠를 가져오는 두 장면
여기서tryLock의 원본 코드를 보면tryAcquire 방법은 자물쇠를 신청하고 자물쇠의 유효기간이 남은 시간을 되돌려주는 것입니다. 자물쇠가 다른 라인에 신청되지 않은 것을 공백으로 설명하기 위해 직접 가져오고 되돌려줍니다. 시간을 얻으면 경쟁 논리에 들어갑니다.

public 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 = this.tryAcquire(leaseTime, unit);
  if(ttl == null) {
   // 
   return true;
  } else {
   // 
  }
 }

경쟁 없음, 직접 잠금 해제
먼저 자물쇠를 확보하고 자물쇠 뒤에 있는 레디스가 무엇을 하고 있는지 확인하고, 레디스의 모니터를 이용하여 백그라운드에서 레디스의 실행 상황을 감시할 수 있다.@RequestLockable을 추가하는 방법을 사용한 후에 사실은 lock과 unlock을 호출하는 것입니다. 다음은 redis 명령입니다.
자물쇠를 채우다
높은 버전의redis는 루아 스크립트를 지원하기 때문에redisson도 이를 지원했고 스크립트 모드를 사용했습니다. 루아 스크립트에 익숙하지 않은 사람은 찾을 수 있습니다.lua 명령을 실행하는 논리는 다음과 같습니다.

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    this.internalLockLeaseTime = unit.toMillis(leaseTime);
    return this.commandExecutor.evalWriteAsync(this.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.singletonList(this.getName()), new Object[]{Long.valueOf(this.internalLockLeaseTime), this.getLockName(threadId)});
  }

잠금 프로세스:

lock키가 존재하는지 판단하고 hset을 호출하여 현재 라인 정보를 저장하고 기한이 지난 시간을 설정하여 nil로 돌아가 클라이언트에게 직접 자물쇠를 가져오라고 알려줍니다

lock 키가 존재하는지 판단하고 존재하는지 판단하면 재입력 횟수를 1을 추가하고 기한이 지난 시간을 다시 설정하여 nil로 돌아가 클라이언트에게 직접 자물쇠를 가져오라고 알려줍니다

다른 라인에 잠겼습니다. 잠금 유효기간의 남은 시간을 되돌려주고 클라이언트에게 기다려야 한다고 알려줍니다

"EVAL" 
"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]);"
 "1" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0" 
 "1000" "346e1eb8-5bfd-4d49-9870-042df402f248:21"

위의 루아 스크립트는 진정한 레디스 명령으로 변환되며, 아래는 루아 스크립트 연산을 거친 후에 실제 실행되는 레디스 명령입니다.

1486642677.053488 [0 lua] "exists" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0"
1486642677.053515 [0 lua] "hset" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0" 
"346e1eb8-5bfd-4d49-9870-042df402f248:21" "1"
1486642677.053540 [0 lua] "pexpire" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0" "1000"

잠금 해제
잠금 해제 프로세스가 복잡해 보입니다.

만약 lock키가 존재하지 않는다면, 자물쇠가 이미 사용 가능하다는 메시지를 보냅니다

자물쇠가 현재 라인에 잠기지 않으면nil로 돌아갑니다

재입력이 가능하기 때문에 잠금 해제 시 재입력 횟수를 1 줄여야 합니다

계산된 재입력 횟수>0이면 만료 시간을 다시 설정합니다

계산된 재입력 횟수<=0이면 자물쇠가 이미 사용 가능하다는 메시지를 보냅니다

"EVAL" 
"if (redis.call('exists', KEYS[1]) == 0) then
 redis.call('publish', KEYS[2], ARGV[1]);
 return 1; end;
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;"
"2" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0" 
"redisson_lock__channel:{lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0}"
 "0" "1000"
 "346e1eb8-5bfd-4d49-9870-042df402f248:21"

경쟁 조건 없이 redis 잠금 해제 명령:
대기 대기열의 라인을 깨우쳐 잠금 해제 메시지를 보냅니다.

1486642678.493691 [0 lua] "exists" "lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0"
1486642678.493712 [0 lua] "publish" "redisson_lock__channel:{lock.com.csp.product.api.service.ProductAppService.searchProductByPage#0}" "0"

경쟁
경쟁이 있는 경우 레디스 측의 루아 스크립트는 같지만 서로 다른 조건이 서로 다른 레디스 명령을 실행하고 복잡한 레디스슨의 원본 코드에 있다.tryAcquire를 통해 자물쇠가 다른 라인에 요청된 것을 발견하면 경쟁 논리에 들어가야 합니다.

this.await는false를 되돌려줍니다. 대기 시간이 자물쇠를 가져오는 최대 대기 시간을 초과했습니다. 구독을 취소하고 자물쇠를 가져오는 데 실패했습니다

this.await는true로 돌아가 순환에 들어가 자물쇠를 가져오려고 시도합니다.

public 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 = this.tryAcquire(leaseTime, unit);
    if(ttl == null) {
      return true;
    } else {
      // 
      time -= System.currentTimeMillis() - current;
      if(time <= 0L) {
        return false;
      } else {
        current = System.currentTimeMillis();
        final RFuture subscribeFuture = this.subscribe(threadId);
        if(!this.await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
          if(!subscribeFuture.cancel(false)) {
            subscribeFuture.addListener(new FutureListener() {
              public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
                if(subscribeFuture.isSuccess()) {
                  RedissonLock.this.unsubscribe(subscribeFuture, threadId);
                }
              }
            });
          }
          return false;
        } else {
          boolean var16;
          try {
            time -= System.currentTimeMillis() - current;
            if(time <= 0L) {
              boolean currentTime1 = false;
              return currentTime1;
            }
            do {
              long currentTime = System.currentTimeMillis();
              ttl = this.tryAcquire(leaseTime, unit);
              if(ttl == null) {
                var16 = true;
                return var16;
              }
              time -= System.currentTimeMillis() - currentTime;
              if(time <= 0L) {
                var16 = false;
                return var16;
              }
              currentTime = System.currentTimeMillis();
              if(ttl.longValue() >= 0L && ttl.longValue() < time) {
                this.getEntry(threadId).getLatch().tryAcquire(ttl.longValue(), TimeUnit.MILLISECONDS);
              } else {
                this.getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
              }
              time -= System.currentTimeMillis() - currentTime;
            } while(time > 0L);
            var16 = false;
          } finally {
            this.unsubscribe(subscribeFuture, threadId);
          }
          return var16;
        }
      }
    }
  }

순환 시도에는 일반적으로 다음과 같은 몇 가지 방법이 있습니다.

while 순환, 한 번에 한 번씩 이어지는 시도, 이 방법의 단점은 대량의 무효한 자물쇠 신청을 초래할 수 있다는 것이다..

Thread.sleep, 위의while 방안에서 수면 시간을 늘려 자물쇠 요청 횟수를 낮추는데 단점은 이 수면 시간 설정이 통제하기 어렵다는 것이다..

정보량을 바탕으로 자물쇠가 다른 자원에 점용될 때 현재 라인 구독 자물쇠의 방출 이벤트는 자물쇠가 방출되면 대기 중인 자물쇠에 대해 메시지를 보내 경쟁을 하고 무효한 자물쇠 신청 상황을 효과적으로 해결한다.핵심 논리는this.getEntry(threadId).getLatch().tryAcquire,this.getEntry(threadId).getLatch () 는 신호량으로 되돌아와 연구할 흥미가 있습니다

redisson 의존
redisson은 자물쇠뿐만 아니라 많은 클라이언트가 redis를 조작하는 방법을 제공하기 때문에 다른 프레임워크, 예를 들어 넷티에 의존한다. 만약에 간단하게 자물쇠를 사용해도 스스로 실현할 수 있다.
이상은 본문의 전체 내용입니다. 본고의 내용이 여러분의 학습이나 업무에 일정한 도움을 줄 수 있는 동시에 저희를 많이 지지해 주시기 바랍니다!