Django 다중 서버 환경 동시성 제어하기(select_for_update/redis lock)

다중 서버 인스턴스 환경에서 하나의 데이터베이스 row에 여러 서버가 동시에 접근할 경우 Race condition이 발생할 수 있습니다.

 

예를 들어, 한정된 수량의 시리얼번호를 요청 순서대로 발급하는 시스템을 구축한다고 가정해 보겠습니다.

이때, 동시에 여러 사용자가 동시에 요청하면 중복 발급이 발생할 가능성이 있습니다.

from django.db import models


class SerialNumber(models.Model):
    id = models.CharField(max_length=32, unique=True)
    is_used = models.BooleanField(default=False)
    created_date = DateTimeField(auto_now_add=True)

다음은 간단한 모델 입니다.

시리얼 번호가 발급되면 is_used 필드를 True로 설정하여 같은 번호는 중복 발급되지 않도록 관리합니다.

 

Race condition이 발생할 수 있는 코드

def get_serial_number():
    serial_number = SerialNumber.objects.filter(is_used=False).first()
    if serial_number:
        serial_number.is_used = True
        serial_number.save(update_fields=["is_used"])
        return serial_number
    return None

위 코드에서는 is_used=False인 시리얼 번호를 조회한 후, 이를 사용 처리(is_used=True)하고 저장합니다. 하지만 다중 요청이 동시에 발생하면 동일한 시리얼 번호를 가져와 중복 발급되는 문제가 발생할 수 있습니다.

 

해결방법 1) select for update를 활용한 row lock

Race condition을 방지하는 한 가지 방법은 데이터베이스 수준에서 row 단위의 lock을 설정하는 것입니다. Django에서는 select_for_update를 활용하여 이를 구현할 수 있습니다.

def get_serial_number():
    from django.db import transaction

    with transaction.atomic():
        serial_number = SerialNumber.objects.select_for_update(
        ).filter(
            is_used=False
        ).first()
    
        if serial_number:
            serial_number.is_used = True
            serial_number.save(update_fields=["is_used"])

        return serial_number

위 예시처럼 트랜잭션 블록 내에서 select_for_update로 row락을 획득하여 시리얼번호를 is_used 상태로 업데이트 할 수 있습니다.

 

1.트랜잭션(transaction.atomic()) 내에서 실행: 트랜잭션이 완료되기 전까지 다른 트랜잭션이 해당 row를 수정할 수 없음

2. select_for_update()를 사용하여 row를 잠금: 하나의 프로세스가 티켓을 가져오면, 다른 프로세스는 해당 row의 락이 해제될 때까지 대기

 

select for update의 문제점

1) 긴 트랜잭션 대기로 인한 부하

select for update는 락이 걸린 row에 다른 트랜잭션의 대기를 유발합니다.

만약 lock을 잡은 트랜잭션 처리가 길어지고 lock 획득을 대기하는 트랜잭션이 많아지면 데이터베이스에 lock 튜플이 증가하며 성능 저하로 이어질 수 있습니다.

 

2) deadlock 발생 가능성

select for update 사용이 잦으면 그만큼 서로 다른 트랜잭션이 서로의 락을 획득하기 위해 무한 대기 상태의 deadlock 발생 가능성이 있어 설계와 사용에 주의해야 합니다.

 

select_for_update(skip_locked=True) 사용

skip locked 옵션을 함께 사용한다면 트랜잭션이 락 획득을 대기하지 않고 락이 없는 다른 row를 선택할 수 있습니다.

def get_serial_number():
    from django.db import transaction

    with transaction.atomic():
        serial_number = SerialNumber.objects.select_for_update(
        skip_locked=True
        ).filter(is_used=False).first()
    
        if serial_number:
            serial_number.is_used = True
            serial_number.save(update_fields=["is_used"])

        return serial_number

 

위와 같이 select_for_update에 skip_locked=True 옵션을 추가해서 시리얼번호가 이미 락이 잡혀있다면 트랜잭션이 대기하지 않고 다른 ticket을 선택하도록 설정할 수 있습니다.

 

• 트랜잭션 대기를 최소화: 이미 락이 걸린 row는 건너뛰므로 빠르게 다른 row를 선택 가능

• Deadlock 위험 감소: 락을 기다리지 않으므로 교착 상태 발생 가능성이 줄어듦

 

하지만 skip_locked 옵션은 mysql 8.0 이상, postgres에서만 지원하기 때문에 사용중인 데이터베이스에 따라 사용이 불가능할 수 있습니다. 또한 요청이 대량으로 발생하면 lock이 걸린 row를 skip하고 추가적인 row를 스캔하기 때문에 DB 부하 발생에 주의해야 합니다.

 

해결방법 2) Redis lock 활용하기

Race Condition 문제를 해결하는 또 다른 방법은 Redis를 활용한 분산 락을 사용하는 것입니다.

Redis는 분산 환경에서 동시성 문제를 해결하기 위한 빠른 락 기능을 제공합니다. 또한 DB 락을 사용하지 않아 DB 부하를 줄일 수 있습니다.

from django.core.cache import cache

def get_serial_number():
    serial_number = SerialNumber.objects.filter(
    is_used=False
    ).first()

    if not serial_number:
    	return None
        
    lock_key = f"lock:serial_number:{serial_number.id}"
    lock = cache.lock(lock_key, timeout=10)
    if lock.acquire(blocking=False):
    	try:
            serial_number.is_used = True
            serial_number.save(update_fields=["is_used"])
            return serial_number
        finally:
        	lock.release()
    return None

django cache를 redis로 설정했다는 가정하에, cache.lock으로 간단하게 redis lock을 구현할 수 있습니다.

key로 고유한 serial_number id를 주어서 row 단위로 lock을 설정할 수 있습니다.

lock.acquire(blocking=False)는 락을 획득하는 메서드인데, blocking=False 옵션으로 이미 락이 존재하는 경우 대기하지 않습니다.

처리가 끝난 후에는 lock.release()로 락을 해제합니다.

 

 

redis lock 내부 구현 살펴보기

조금 더 들어가서 python redis lock 내부 구현을 한번 살펴보겠습니다.

아래와 같이 redis lock은 하나의 클래스로 만약 위 로직의 lock = cache.lock(lock_key, timeout=10) 부분은 레디스 클라이언트 하위의 lock 클래스 인스턴스를 생성하는 것으로 볼 수 있습니다.

redis/lock.py

class Lock:
    def __init__(
        self,
        redis,
        name: str,
        timeout: Optional[Number] = None,
        sleep: Number = 0.1,
        blocking: bool = True,
        blocking_timeout: Optional[Number] = None,
        thread_local: bool = True,
    ):
    
    ...

    def acquire(
    self,
    sleep: Optional[Number] = None,
    blocking: Optional[bool] = None,
    blocking_timeout: Optional[Number] = None,
    token: Optional[str] = None,
):
    if sleep is None:
        sleep = self.sleep
    if token is None:
        token = uuid.uuid1().hex.encode()
    else:
        encoder = self.redis.get_encoder()
        token = encoder.encode(token)
    if blocking is None:
        blocking = self.blocking
    if blocking_timeout is None:
        blocking_timeout = self.blocking_timeout
    stop_trying_at = None
    if blocking_timeout is not None:
        stop_trying_at = mod_time.monotonic() + blocking_timeout
    while True:
        if self.do_acquire(token):
            self.local.token = token
            return True
        if not blocking:
            return False
        next_try_at = mod_time.monotonic() + sleep
        if stop_trying_at is not None and next_try_at > stop_trying_at:
            return False
        mod_time.sleep(sleep)
        
    def do_acquire(self, token: str) -> bool:
        if self.timeout:
            # convert to milliseconds
            timeout = int(self.timeout * 1000)
        else:
            timeout = None
        if self.redis.set(self.name, token, nx=True, px=timeout):
            return True
        return False
        
    def release(self) -> None:
        expected_token = self.local.token
        if expected_token is None:
            raise LockError("Cannot release an unlocked lock", lock_name=self.name)
        self.local.token = None
        self.do_release(expected_token)

    def do_release(self, expected_token: str) -> None:
        if not bool(
            self.lua_release(keys=[self.name], args=[expected_token], client=self.redis)
        ):
            raise LockNotOwnedError(
                "Cannot release a lock that's no longer owned",
                lock_name=self.name,
            )

락을 획득하기 위한 acquire 메서드는 무한 루프 내에서 대기하며 락 획득을 시도하는데 이 때 획득이 실패했을 경우 blocking=False 이면 대기하지 않고 메서드를 종료합니다.

실제로 redis에 lock key를 고유 토큰과 함께 set 하는 do_acquire 메서드에서는 nx=True 인자가 존재하는데, 이는 해당 키가 존재하지 않을 경우 셋하는 SET NX 명령어로 동작하게 됩니다. px는 타임아웃 설정입니다.

 

락을 해제하는 release 메서드에서는 내부적으로 아래와 같이 key를 통해 가져온 value가 생성한 토큰과 일치하는지 검증 후 일치하면 삭제해주는 스크립트를 실행합니다. 

LUA_RELEASE_SCRIPT = """
    local token = redis.call('get', KEYS[1])
    if not token or token ~= ARGV[1] then
        return 0
    end
    redis.call('del', KEYS[1])
    return 1
"""

 

Redis lock 재시도 로직 추가하기

def get_serial_number():
    max_retries = 5
    exclude_ids = []
    for _ in range(max_retries):
        serial_number = SerialNumber.objects.filter(is_used=False).exclude(id__in=exclude_ids).order_by("id").first()
        if not serial_number:
            return None

        lock_key = f"lock:serial_number:{serial_number.id}"
        lock = cache.lock(lock_key, timeout=10)
        if lock.locked(lock_key):
            exclude_ids.append(serial_number.id)
            continue

        if lock.acquire(blocking=False):
            try:
                do_something()
                serial_number.is_used = True
                serial_number.save(update_fields=["is_used"])
                return serial_number
            finally:
                lock.release()
    return None

불가피하게 트랜잭션 내 처리가 다소 길 경우 (위 예제에서는 do_something 메서드로 긴 트랜잭션을 표현했습니다.)

이미 락이 잡혀있는 시리얼 번호가 다른 프로세스의 요청에서 반복적으로 획득 시도될 가능성이 존재합니다.

이러한 경우 최대 시도 횟수(max_retries) 범위 내에서 이미 락이 잡혀있는 key를 cache.locked()메서드로 체크합니다.

 

lock.locked는 인수로 받은 key가 이미 존재하는지 체크만 해주는 메서드로 django orm의 exists()와 유사하게 동작합니다.

따라서 해당 시리얼번호가 이미 락이 획득된 상태일 경우 해당 id는 제외하고 재조회를 시도하여 보다 락 획득 실패율을 낮출 수 있습니다.