Java,MySQL,Redis로 동시성 문제 해결하기

재고시스템으로 알아보는 동시성 이슈 해결방법을 듣고 정리한 내용입니다.

    @Test
    public void 동시에_50명() throws InterruptedException {
		// 처음 Quantity를 100으로 등록함. 
		Stock stock = new Stock(1L, 100L);
        stockRepository.saveAndFlush(stock);
        
        int threadCount = 100;
        // 비동기로 실행
        ExecutorService executorService = Executors.newFixedThreadPool(32);
        // 요청이 끝날 때까지 기다리는 용도
        CountDownLatch latch = new CountDownLatch(threadCount);

        for (int i = 0; i < threadCount; i++) {
            executorService.submit(() -> {
                try {
                    stockService.decrease(1L, 1L);
                } finally {
                    latch.countDown();
                }
            });
        }

        latch.await();

        Stock stock = stockRepository.findById(1L).orElseThrow();

        // threadCount가 100이므로 100번 decrease하면 수량은 0이 되어야한다.
        assertEquals(0, stock.getQuantity());
    }

decrease 메서드는 단순히 아래처럼 구현되어있다.

public void decrease(Long quantity) {
        if (this.quantity - quantity < 0) {
            throw new RuntimeException("foo");
        }

        this.quantity -= quantity;
    }

결과는? 실패한다.
왜냐하면 Race Condition이 일어났기 때문이다.
첫번째 스레드가 데이터를 가져가서 갱신하고, 그 다음 두번째 스레드가 데이터를 가져가서 갱신하는 것이 아닌,
첫번째 스레드가 데이터를 가져가서 갱신하는 동안 두번째 스레드가 공유 데이터(quantity)에 접근해 갱신하기 때문에, 문제가 발생하는 것이다.

예를 들어, quatity가 예제 코드처럼 100이었다면 Thread#1이 가져가서 99로 바꾸는 와중 (아직까지 quantity는 100인 상태) Thread#2도 그에 접근해 decrease를 수행한다면 순차적으로 진행되었을 땐 98이 되어야할 값이 둘 다 100 -> 99로 바꾸는 것을 수행하기 때문에 문제가 발생하는 것!

Solution 1. Synchronized

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public synchronized void decrease(Long id, Long quantity) {
        Stock stock = stockRepository.findById(id).orElseThrow();
        stock.decrease(quantity);
        stockRepository.saveAndFlush(stock);
    }

가장 먼저 decrease 메서드에 synchronized 키워드를 붙여보자. synchronized는 하나의 스레드만 해당 메서드에 접근할 수 있도록 해 준다. 다만 문제는 해결되지 않는다.

Why? 스프링에서는 @Transactional을 사용하게 되면 우리가 만든 클래스를 래핑해서 실행하게 된다. (Proxy로 감싸서 AOP가 동작할 수 있도록 한다.) 즉, 해당 메서드의 시작과 끝에 TX가 실행되고 종료될 수 있도록 한다.

실제 데이터베이스에 업데이트하기 전에 proxy에서 decrease에 접근할 수 있기 때문에 (갱신되기 이전 값에 접근 가능하여) 문제가 발생하는 것.

만약 @Transactional 애노테이션을 제거하고 테스트케이스를 실행한다면 정상적으로 동작할 것이다.

그럼 Synchronized를 사용했을 때 문제는 없을까?
하나의 프로세스 안에서만 보장되는 키워드이기 때문에 서버가 한대일 땐 괜찮지만, 서버가 여러 대일 경우에는 데이터에 대한 접근을 여러 대에서 할 수 있게 된다.

Solution 2. Using Database

• Pessimistic Lock
  실제로 데이터에 Lock을 걸어서 정합성을 맞추는 방법. exclusive lock을 걸게되며 다른 트랜잭션에서는 lock이 해제되기전에 데이터를 가져갈 수 없다. 단, 데드락이 발생할 수 있다.
  Row나 Table 단위로 락이 걸림.
  

@Lock(value = LockModeType.PESSIMISTIC_WRITE)
@Query("select s from Stock s where s.id=:id")
Stock findByIdWithPessimisticLock(Long id);

• Optimistic Lock
  실제로 Lock 을 이용하지 않고 버전을 이용함으로써 정합성을 맞추는 방법. 먼저 데이터를 읽은 후에 update 를 수행할 때 현재 내가 읽은 버전이 맞는지 확인하며 업데이트. 내가 읽은 버전에서 수정사항이 생겼을 경우에는 application에서 다시 읽은 후에 작업을 수행해야 한다.
  예를 들어 Sever#1이 ver#1인 값 {100}을 읽어와서 {99}로 수정할 때, 조건문에 ver#1을 명시해서 실행하기 때문에 ver#2로 업데이트가 된다. 근데 만약 동시에 ver#1인 {100}에 Server#2에서 접근해서 {99}로 업데이트하려고 한다면, 실제 DB에 반영되어있는 값은 가장 최신 값이 ver#2이기 때문에 Server#2의 작업은 실패하게 된다.

@Lock(value = LockModeType.OPTIMISTIC)
@Query("select s from Stock s where s.id = :id")
Stock findByIdWithOptimisticLock(Long id);`

Stock 엔티티에는 version을 명시해주어야한다.
@Version
private Long version;

별도의 락을 잡지않아 Pessimistic Lock보다 성능이 좋지만, 업데이트에 실패한 경우에 대한 처리를 개발자가 직접 해줘야함. 충돌이 빈번히 일어나지 않는 경우에 적합하다.

• Named Lock
  이름을 가진 metadata locking. 이름을 가진 lock 을 획득한 후 해제할때까지 다른 세션은 이 lock 을 획득할 수 없도록 한다. 주의할점으로는 transaction이 종료될 때 lock 이 자동으로 해제되지 않는다. 별도의 명령어로 해제를 수행해주거나 선점시간이 끝나야 해제된다.
  메타데이터를 기준으로 락이 걸림.

public interface LockRepository extends JpaRepository<Stock, Long> {
    @Query(value = "select get_lock(:key, 3000)", nativeQuery = true)
    void getLock(String key);

    @Query(value = "select release_lock(:key)", nativeQuery = true)
    void releaseLock(String key);
}

락을 사용하는 경우 실무에서는 데이터 소스를 분리하는 것이 낫다. 커넥션 풀이 부족해질 수 있기 때문.

위의 LockRepository를 decrease의 호출 전과 후에 getLock, releaseLock을 해줌으로써 사용할 수 있다.

Solution 3. Using Redis

• Lettuce
  setnx 명령을 활용해 분산락 구현 가능. 기존에 등록된 값이 없을 때만 set하는 기능. spin lock 방식이므로 retry 로직을 개발자가 작성해줘야함. lock 획득을 원하는 곳에서 주기적으로 lock 획득이 가능할 때가지 재시도하는 방법을 spin lock이라고 한다.
  

    public Boolean lock(Long key) {
        return redisTemplate
                .opsForValue()
                .setIfAbsent(generateKey(key), "lock", Duration.ofMillis(3_000));
    }

    public Boolean unlock(Long key) {
        return redisTemplate.delete(generateKey(key));
    }

    private String generateKey(Long key) {
        return key.toString();
    }

구현이 간단하지만 spin-lock 방식이므로 레디스에 부하를 줄 수 있어서, sleep을 활용해 락 획득 - 재시도 간에 텀을 두어야한다.
재시도가 필요하지 않은 lock은 lettuce를 활용하는 것이 나을 수 있다.

• Redisson
  pub-sub 기반으로 Lock 구현. 채널 하나를 만들고 락 획득하려고 대기중에게 락 점유가 가능하다고 알려주는 방법.
  

    public void decrease(Long key, Long quantity) {
        RLock lock = redissonClient.getLock(key.toString());

        try {
            boolean available = lock.tryLock(10, 1, TimeUnit.SECONDS);

            if (!available) {
                System.out.println("lock 획득 실패");
                return;
            }

            stockService.decrease(key, quantity);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

Redisson을 활용하려면 별도의 라이브러리를 활용해줘야한다.
재시도가 필요한 경우에는 redisson을 활용하는 것이 나을 수 있다.