Redis로 구현하는 분산 세마포어와 카운트다운 래치

궁금하면 500원·2024년 6월 7일

CountDownLatch Java Semaphore redis redisson 개발공부 동시성제어 분산락 분산시스템 소스코드

미생의 개발 이야기

목록 보기

5/60

분산 환경에서의 Redisson 동기화 컴포넌트: Semaphore와 CountDownLatch

서론

이번 시간에는 Redisson의 분산 동기화 컴포넌트를 계속해서 살펴보겠습니다.
특히, Semaphore(세마포어)와 CountDownLatch(카운트다운 래치)가 어떻게 Redis를 활용하여 분산 환경에서도 일관성을 보장하며 작동하는지 집중적으로 알아보겠습니다.

일반적인 자바 스레드 프로그래밍에서 Semaphore는 특정 자원에 동시에 접근할 수 있는 스레드 수를 제한하는 데 사용됩니다.
반면 CountDownLatch는 다른 스레드들의 작업이 완료될 때까지 기다리는 용도로 사용되죠.
하지만 애플리케이션이 분산 환경에 배포되면, 단일 JVM에 국한된 이 객체들은 노드를 넘어선 협업 제어가 불가능합니다.
이때 Redisson이 제공하는 RSemaphore와 RCountDownLatch를 활용하면, 분산 환경에서도 이 두 가지 동기화 패턴을 효과적으로 사용할 수 있습니다.

Redisson 컴포넌트 소개

1. RSemaphore

주요 사용 시나리오:

데이터베이스 연결, 캐시 접근, 특정 핵심 로직 등 공유 자원에 동시에 접근하는 병렬 작업 수를 제한할 때 사용합니다.

핵심 특징:

가용 허가(permit) 개수를 동적으로 조정할 수 있습니다.
분산 클러스터 환경에서 모든 노드가 남아있는 허가 개수를 정확하게 인지할 수 있습니다.

2. RCountDownLatch

주요 사용 시나리오:

여러 개의 하위 작업이 완료될 때까지 메인 프로세스가 대기해야 하는 "카운트다운"과 유사한 상황에 적합합니다.
예를 들어, 분산 시스템의 여러 서브태스크가 모두 완료되어야만 주 프로세스가 다음 단계로 넘어갈 수 있는 경우입니다.

핵심 특징:

초기 카운트다운 값을 설정할 수 있으며, 각 노드가 작업 진행에 따라 카운트를 줄일 수 있습니다.
카운트가 0이 되면, 해당 래치를 기다리고 있던 모든 노드들을 깨웁니다.

이제 이 두 컴포넌트가 어떻게 구현되었는지 자세히 살펴보겠습니다.

Semaphore 구현 방식

Redisson의 분산형 세마포어 인터페이스인 RSemaphore는 내부적으로 Redis의 Lua 스크립트와 key의 증감 기능을 사용합니다.

허가 획득(acquire):
- Lua 스크립트가 Redis에 저장된 현재 가용 허가 개수(키에 기록된 값)를 확인합니다.
- 허가 개수가 충분하면 값을 감소시키고, 그렇지 않으면 스레드를 블록하거나 실패를 반환합니다.
허가 해제(release):
- Redis에 저장된 허가 개수를 1 증가시키고, 블록되어 대기 중인 스레드(존재하는 경우)에게 알립니다.

핵심 구조

RedissonSemaphore 클래스는 주로 비동기 메서드(tryAcquireAsync, releaseAsync 등)를 통해 구현됩니다. 이 메서드들은 최종적으로 Lua 스크립트를 생성하여 Redis로 보내고, Redis가 원자적으로 증감 및 블록 해제 알림을 처리하도록 합니다.

public class RedissonSemaphore extends RedissonBaseDistributedObject implements RSemaphore {
    // ...
    @Override
    public void acquire() throws InterruptedException {
        tryAcquire(-1, TimeUnit.MILLISECONDS);
    }
    @Override
    public boolean tryAcquire(long waitTime, TimeUnit unit) throws InterruptedException {
        return get(tryAcquireAsync(waitTime, unit, 1));
    }
    @Override
    public void release() {
        get(releaseAsync(1));
    }
    // ...
}

허가 획득 구현 방식:
tryAcquireAsync 내부에서는 Lua 스크립트를 통해 Redis의 가용 허가 개수가 충분한지 먼저 확인합니다.
만약 충분하다면 DECRBY 명령어로 허가 수를 감소시킵니다.
허가 수가 부족하면 지정된 대기 시간과 모드에 따라 스레드를 블록하거나 false를 반환합니다.

```java
private RFuture<Boolean> tryAcquireAsync(long waitTime, TimeUnit unit, int permits) {
    // Lua 스크립트 핵심: 키의 값이 permits보다 크거나 같은지 판단 후, 
    // 충분하면 decrby 실행, 아니면 0 반환
    return commandExecutor.evalWriteAsync(
      getName(), 
      RedisCommands.EVAL_BOOLEAN, 
      "local currVal = redis.call('get', KEYS[1]); " + 
      "if (currVal ~= false and tonumber(currVal) >= tonumber(ARGV[1])) then " +
          "redis.call('decrby', KEYS[1], ARGV[1]); " + 
          "return 1; " +
      "end; " +
      "return 0;", 
      Collections.<Object>singletonList(getName()),
      permits);
}
```

허가 해제 구현 방식:
release는 획득과 반대로, Redis에 저장된 허가 개수를 증가시킵니다.
이때 INCRBY 명령어를 사용하며, 허가 개수가 0에서 양수로 바뀌면 대기 중인 스레드에게 알리는 로직이 추가됩니다.

CountDownLatch 구현 방식

Redisson의 RCountDownLatch도 Semaphore와 유사하게 Redis를 사용하여 "카운트다운 값"을 유지합니다.
이 값이 0보다 크면 await()을 호출한 스레드는 블록되고, 0이 되면 모든 블록된 스레드가 깨어납니다.

핵심 구조

RedissonCountDownLatch 클래스에는 다음과 같은 핵심 메서드들이 있습니다.

public class RedissonCountDownLatch extends RedissonBaseDistributedObject implements RCountDownLatch {
    // ...
    @Override
    public void await() throws InterruptedException {
        get(awaitAsync());
    }
    @Override
    public void countDown() {
        get(countDownAsync());
    }
    @Override
    public boolean trySetCount(long count) {
        return get(trySetCountAsync(count));
    }
    // ...
}

trySetCount: 카운트다운 값을 초기화하거나 재설정합니다.
countDown: 카운트 값을 1 감소시킵니다.
await: 카운트가 0이 될 때까지 블록 상태로 대기합니다.
카운트 초기화 구현 방식:
trySetCount는 Redis에 카운트 값을 설정합니다.
만약 해당 키가 이미 존재하면 실패하여, 이미 카운트다운이 진행 중임을 알려줍니다.

카운트 감소 구현 방식:
countDown의 Lua 스크립트는 Redis의 DECR 명령어를 사용하여 카운트를 감소시킵니다.
이때, 만약 카운트가 0이 되면 PUBLISH 명령을 통해 특정 채널에 메시지를 발행하여, await() 중인 스레드를 깨우는 역할을 합니다.

```java
private RFuture<Long> countDownAsync() {
    return commandExecutor.evalWriteAsync(
        getName(),
        RedisCommands.EVAL_LONG,
        "local val = redis.call('decr', KEYS[1]); " + 
        // 카운트가 0이 되면, 특정 채널에 메시지를 발행하여 대기 중인 스레드를 깨움
        "if val <= 0 then " +
            "redis.call('publish', KEYS[2], '0'); " +
        "end;" +
        "return val;",
        Arrays.asList(getName(), getChannelName(getName()))
    );
}
```

대기(await) 구현 방식:
분산 환경에서 await는 Redis의 Pub/Sub (발행/구독) 메커니즘을 구독하는 방식으로 구현됩니다.
카운트가 0이 되어 서버 측에서 PUBLISH가 실행되면, 클라이언트가 이 메시지를 수신하여 블록 상태에서 해제됩니다.

일관성과 원자성 보장

이전에 살펴본 분산 락과 마찬가지로, Semaphore와 CountDownLatch의 구현은 Redis의 원자적 연산과 Lua 스크립트에 의존합니다.

원자적 연산: INCRBY, DECRBY, PUBLISH와 같은 Redis 명령어들은 서버 측에서 한 번에 완료되므로, 동시성 문제를 방지합니다.
Lua 스크립트: '허가 개수가 충분한지 판단 -> 감소'와 같은 복잡한 로직은 Lua 스크립트로 묶어 실행함으로써 불가분성을 보장합니다.
Pub/Sub: 허가 개수나 카운트 값에 중요한 변화가 발생하면, Redis의 Pub/Sub 채널을 통해 다른 노드에 있는 블록된 스레드에게 알림을 보냅니다.

이러한 특성을 활용하여, Redisson은 한 JVM에서 실행된 tryAcquire()나 countDown() 같은 작업이 원격 Redis의 데이터를 변경하고, 그 결과를 다른 JVM에 있는 대기 중인 스레드에게 동기화함으로써 분산 환경에서의 협업을 가능하게 만듭니다.