Open in app

Sign in

Write

Sign in

Weekly Java: 간단한 재고 시스템으로 학습하는 동시성 이슈

Synchronized vs Pessimistic Lock vs Optimistic Lock vs Distributed Lock

Sigrid Jin
41 min readAug 21, 2022

이번 시간에는 간단한 재고 시스템을 만들어보면서 동시성 이슈를 해결하는 방법에 대해 함께 고민해보도록 하자. 개인적으로 인프런에 올라온 재고시스템으로 알아보는 동시성이슈 해결방법을 완강하고 작성하는 노트이다. 강의를 추천하니 꼭 들으시기를 바란다.

백엔드 개발자에게 동시성 이슈는 중요하다. 프로젝트를 시작할 때 동시성 이슈를 고려하지 않고 개발을 시작하게 되면 데이터 정합성이 중요한 상황에서 여러 문제가 발생할 수 있기 때문이다. 우리가 일반적으로 이야기하는 Race Condition은 둘 이상의 스레드가 공유 데이터에 액세스할 수 있고, 동시에 변경하려고 할 때 발생하는 문제이다. 예를 들어, Stock이라는 클래스가 있고 Quantity라고 하는 프로퍼티가 있다고 생각해보자.

이 때, 웹은 기본적으로 멀티 쓰레드 환경이기 때문에 두 개 이상의 쓰레드가 특정 값을 읽거나 쓰려고 할 것이다. 문제가 발생하는 경우는 두 쓰레드가 동시에 업데이트를 하려는 경우이다. 쓰레드 A와 쓰레드 B가 각각 Quantity 프로퍼티를 불러와서 업데이트를 하고자 하는데, 상대방이 업데이트 하고자 하는 값을 알 수 없기 때문에 값이 의도한 대로 업데이트되지 않는다. 이 때 하나의 쓰레드만 데이터에 액세스할 수 있도록 허용하는 것이 기본적인 해결책이다.

다음 요구 사항을 구현해보면서 동시성 이슈를 해결하는 방법에 대해 살펴보도록 하자.

요구 사항

Quantity 변수의 기존 값은 1000이었다. 이를 1000개의 쓰레드가 각각 개별로 1씩 감소하여 0으로 만드는 테스트를 작성하고, 이를 통과시키는 비즈니스 로직을 작성하여라.

각종 링크

  1. 인프런 강의
  2. GitHub 저장소

테스트 환경

  • Apple Macbook Pro M1
  • Docker Compose-based MySQL
  • Docker Compose-based Redis
  • JDK 11 & Spring Data JPA
  • IntelliJ & JUnit 5

테스트 1: 동시성을 고려하지 않은 기본적인 로직

일단 동시성을 고려하지 않으면 다음과 같이 구현할 수 있다.

JPA 환경에서 구동하는 프로젝트이기 때문에, 트랜잭션 시작과 커밋 그리고 롤백 기능을 @Transactional 어노테이션으로 메소드 레벨에서 갈음하였다.

 // StockService.java @Transactional
 public void decreaseV1(final Long productId, final Long quantity) {
  final Stock stock = stockRepository.getByProductId(productId);
  stock.decrease(quantity);
 }

1개의 쓰레드만 존재할 때는 우리의 예상대로 잘 동작할 것이다.

 // StockServiceTest.java private final int threadCount = 300;
 private final long productId = 10L;
 private final long quantity = 1L;
 private final long initQuantity = 300L; private ExecutorService executorService;
 private CountDownLatch countDownLatch;@BeforeEach
 public void beforeEach() {
  stockRepository.save(new Stock(productId, initQuantity)); executorService = Executors.newFixedThreadPool(threadCount);
  countDownLatch = new CountDownLatch(threadCount);
 } @AfterEach
 public void afterEach() {
  stockRepository.deleteAll();
 } @DisplayName("단일 쓰레드일 때를 테스트한다")
 @Test
 void 단일_쓰레드로_재고를_감소시킨다() {
  // given  // when
  stockService.decreaseV1(productId, quantity);  // then
  final long afterQuantity = stockRepository.getByProductId(productId).getQuantity();
  assertThat(afterQuantity).isEqualTo(initQuantity - 1);
 }

하지만 다음과 같이 테스트를 구현할 경우 멀티 쓰레드 환경에서 쓰레드 간의 충돌이 일어나 절대 0 로의 완전한 뺄셈이 이루어질 수 없다. 실제로 테스트를 구동할 경우, 테스트를 동작시킬 때마다 output이 어떨 때는 50, 어떨 때는 30, 어땔 때는 70.. 이렇게 서로 상이함을 확인할 수 있다.

// StockServiceTest.java
    @Test
    void 멀티_쓰레드를_사용한_재고_감소() throws InterruptedException {
        // given        // when
        IntStream.range(0, threadCount).forEach(e -> executorService.submit(() -> {
                    try {
                        stockService.decrease(productId, quantity);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
        ));        countDownLatch.await();        // then
        final Long afterQuantity = stockRepository.getByProductId(productId).getQuantity();
        System.out.println("### 동시성 처리 이후 수량 ###" + afterQuantity);
        assertThat(afterQuantity).isZero();
    }

테스트 2: Synchronized 키워드를 적용한 동기화 로직

이제 Java 진영에서 동시성 처리할 때 유명한 synchronized 키워드를 이용해보자. 해당 키워드를 이용하면 현재 접근하고 있는 메소드에 일정 쓰레드에 락을 거는 전통적인 접근을 취할 수 있다. 다만 Synchronized 키워드는 내부적으로 모든 동작에 대해 Lock을 걸기 때문에, 필요한 부분만 Lock을 거는 다른 기법들에 비해 성능상 오버헤드가 심하다는 평가를 받는다.

public synchronized void decrease(final Long id, final Long quantity) {
    // 1. get stock
    // 2. decrease stock
    // 3. save stock

    final Stock stock = stockRepository.getByProductId(id);
    stock.decrease(quantity);
    stockRepository.saveAndFlush(stock);
}

이번 접근에는 @Transactional 이라는 어노테이션을 제외한다. 우리가 Spring Data 제품군을 활용할 때 @Transactional이라는 어노테이션을 붙이는 이유는, 프로젝트가 커지고 DAO, Service가 많으면 많아질수록 중복되는 if, try catch 코드가 점점 많아지기 때문에 중복 코드는 한 군데에서 관리하고 싶다는 요구사항이 발생했기 때문이리라.

이 때 @Transactional 어노테이션을 붙이면 그 try catch 코드들을 직접 짤 필요 없이 알아서 자동으로 붙여주기 때문에 중복되는 코드가 줄어들고 보기 쉬워진다. 우리의 이번 키워드에서는 @Transactional 어노테이션을 사용하지 않았다. 해당 어노테이션의 사용은 불필요한 Race Condition을 유발시킬 수 있기 때문이다.

트랜잭션은 시작되었으면 커밋이 되어야 하고, 커밋에 실패한 트랜잭션으로 인한 DB 변화가 존재한다면 롤백을 시켜야 한다. 따라서 특정 쓰레드의 작업 도중에 다른 쓰레드가 공유 자원을 읽어와서 생길 수 있는 이슈를 방지하려고 한다. 따라서 실질적으로 비즈니스 로직을 구현해줄 때는 Repository에서 save와 flush를 모두 수동으로 진행해주어야 한다.

// StockServiceTest.java@DisplayName("SYNCHRONIZED를 사용한 재고 감소 - 동시 1000개 테스트 | 16.994s 소요")
    @Test
    void SYNCHRONIZED를_사용한_재고_감소() throws InterruptedException {
        // given// when
        IntStream.range(0, threadCount).forEach(e -> executorService.submit(() -> {
                    try {
                        stockService.decrease(productId, quantity);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
        ));countDownLatch.await();// then
        final Long afterQuantity = stockRepository.getByProductId(productId).getQuantity();
        System.out.println("### SYNCHRONIZED 동시성 처리 이후 수량 ###" + afterQuantity);
        assertThat(afterQuantity).isZero();
    }

이제 다음 테스트를 실행하면 통과할 것이다. 하지만 큰 문제가 하나 있다. 우리는 Method Level에 Synchronized 를 사용하였으나, 해당 키워드는 같은 프로세스 단위에서만 동시성을 보장한다. 따라서 서버가 1대일 때는 동시성 이슈가 해결되는 듯 하나, 여러 대의 서버를 활용하면 여러 개의 인스턴스가 존재하는 것과 동일하기 때문에 실질적으로 웹 환경에서는 동시성을 보장하지 못한다.

테스트 3: Pessimistic Lock의 사용

Pessimistic Lock이란 실제로 데이터에 Lock을 걸어서 정합성을 맞추는 방법이다. 이번 기법에서 Exclusive Lock이라는 것을 적용하게 되면 다른 트랜잭션에서는 Lock이 해제되기 전까지 데이터를 가져갈 수 없으므로 데이터 정합성을 보장하게 된다.

구체적으로는, Method Level에 @Transactional 및 DB 조회 시에 @Lock(LockModeType.PESSIMISTIC_WRITE) 을 사용하여 트랜잭션이 시작할 때 Shared/Exclusive Lock을 적용하게 된다. 따라서 Pessimistic Lock은 동시성 충돌이 잦을 것으로 예상되어 동시성을 강력하게 지켜야 할 때 사용하여야 한다.

어떨 때 사용하면 좋을까. 충돌이 빈번하게 일어난다면 Optimistic Lock보다 성능 좋고, 데이터 정합성이 안정적이다. 하지만 별도의 Lock을 잡기 때문에 속도가 느리고, 경우에 따라 Dead Lock의 위험성이 있음은 유의해야 한다.

JPA 환경에서는 다음과 같이 Repository를 선언해주면 MySQL에서 기본 옵션으로 제공하는 Pessimistic Lock을 사용할 수 있게 된다.

// PessimisticStockRepository.javapublic interface PessimisticStockRepository extends JpaRepository<Stock, Long> {

    @Lock(LockModeType.PESSIMISTIC_WRITE)
    Stock getByProductId(Long productId);
}

이제 서비스 객체를 구현한다. 1번 방법처럼 @Transactional 어노테이션을 활용한다.

// PessimisticLockStockService.java@Service
public class PessimisticLockStockService implements StockBusinessInterface {private PessimisticStockRepository stockRepository;public PessimisticLockStockService(final PessimisticStockRepository stockRepository) {
        this.stockRepository = stockRepository;
    }    @Transactional
    public void decrease(final Long id, final Long quantity) {
        Stock stock = stockRepository.getByProductId(id);
        stock.decrease(quantity);
    }
}

이제 테스트를 작성하자. 우리는 executorService를 사용하여 newFixedThreadPool를 만들어 쓰레드 갯수만큼 초기화해둔다. 그리고, CountDownLatch를 활용하여 임의의 쓰레드가 다른 쓰레드의 작업이 종료될 때까지 기다리도록 만들 수 있다.

CountDownLatch는 쓰레드를 N개 실행했을 때, 일정 개수의 쓰레드가 모두 끝날 때 까지 기다려야만 다음으로 진행할 수 있거나 다른 쓰레드를 실행시킬 수 있는 경우 사용한다. 예를 들어 리스트에 어떤 자료구조가 있고, 각 자료구조를 병렬로 처리한 후 배치(batch)로 데이터베이스를 업데이트 한다거나 다른 시스템으로 push하는 경우를 생각해볼 수 있겠다.

Tip. CountDownLatch의 어떤 점이 이를 가능하게 하는가?

1. CountDownLatch를 초기화 할 때 정수값 count를 넣어준다.
2. 쓰레드는 마지막에서 countDown() 메서드를 불러준다.
3. 그러면 초기화 때 넣어준 정수값이 하나 내려간다.
4.즉 각 쓰레드는 마지막에서 자신이 실행 완료했음을 countDown 메서드로 알려준다.
5. 이 쓰레드들이 끝나기를 기다리는 쪽 입장에서는 await()메서드를 불러준다.
6. 그러면 현재 메서드가 실행중이 메인 쓰레드는 더이상 진행하지않고 CountDownLatch의 count가 0이 될 때까지 기다린다. (CountDownLatch.await() 메소드)
7. 0이라는 정수값이 게이트(Latch)의 역할을 한다. 카운트다운이 되면 게이트(latch)가 열리는 것이다.

이제 다음 테스트를 실행하면 통과할 것이다. 본인의 컴퓨터에는 결과값이 출력되기까지 약 12.415초가 소요되었다.

@DisplayName("pessimistic lock을 사용한 재고 감소 - 동시에 1000개 테스트 | 12.415s 소요")
    @Test
    void PESSIMISTIC_LOCK을_사용한_재고_감소() throws InterruptedException {
        // given// when
        IntStream.range(0, threadCount).forEach(e -> executorService.submit(() -> {
                    try {
                        pessimisticLockStockService.decrease(productId, quantity);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
        ));countDownLatch.await();// then
        final Long afterQuantity = stockRepository.getByProductId(productId).getQuantity();
        System.out.println("### PESSIMISTIC LOCK 동시성 처리 이후 수량 ###" + afterQuantity);
        assertThat(afterQuantity).isZero();
    }

테스트 4. Optimistic Lock의 사용

Optimistic Lock은 실제로 Lock을 사용하지 않고 Version을 이용함으로써 데이터 정합성을 준수하는 방법이다. Method Level에 @Transactional 및 DB 조회시 @Lock(LockModeType.OPTIMISTIC) 를 사용하게 된다.

먼저 데이터를 읽은 후에 update를 실행하고, 이 때 현재 내가 읽은 버전이 맞는 지 확인하는 Query를 조회한다. 다음, 내가 읽은 Version에서 수정사항이 생겨서 Version의 값이 증가하였다면, 새롭게 App에서 데이터를 다시 읽은 후에 작업을 수행해야 한다.

추가 기능을 별도로 구현해야 하는 번거러움이 존재한다. 먼저, Version 관리를 위하여 테이블을 마이그레이션 하여야 한다. 또한, Version 충돌 시 재시도 로직을 구현해야 한다. 마지막으로, DB 기본 트랜잭션을 활용하지 않기 때문에 롤백을 직접 구현해야 한다. 실제로 이번 테스트 케이스에서는 Version 충돌이 많기에 Optimistic Lock의 성능이 가장 좋지 않다.

먼저, 다음과 같이 Repository를 선언하면 MySQL이 기본으로 제공하는 Optimistic Lock을 사용할 수 있다.

public interface OptimisticStockRepository extends JpaRepository<Stock, Long> {
    @Lock(LockModeType.OPTIMISTIC)
    Stock getByProductId(Long productId);
}

도메인 객체에 Version 프로퍼티를 추가하여 현재 도메인 객체의 최신 Version 값을 추적할 수 있도록 리팩토링한다.

// Stock.javapackage com.example.stock.domain;import javax.persistence.Version;
import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;@Entity
public class Stock {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;private Long productId;private Long quantity;public Stock() {}public Stock(Long productId, Long quantity) {
        this.productId = productId;
        this.quantity = quantity;
    }@Version
    private Long version;public Long getQuantity() {
        return quantity;
    }public Long decrease(Long quantity) {
        if (this.quantity < quantity) {
            throw new IllegalArgumentException("Not enough stock");
        }this.quantity -= quantity;
        return this.quantity;
    }
}

서비스 객체는 기존의 서비스와 차이가 없다.

package com.example.stock.service;import com.example.stock.domain.Stock;
import com.example.stock.repository.OptimisticStockRepository;
import com.example.stock.repository.StockRepository;
import org.springframework.stereotype.Service;import javax.transaction.Transactional;@Service
public class OptimisticLockStockService implements StockBusinessInterface {private final OptimisticStockRepository stockRepository;public OptimisticLockStockService(final OptimisticStockRepository stockRepository) {
        this.stockRepository = stockRepository;
    }    @Transactional
    public void decrease(final Long id, final Long quantity) {
        final Stock stock = stockRepository.getByProductId(id);
        stock.decrease(quantity);
    }
}

이제 다음 코드를 살펴보자. 지속적으로 DB 변경을 재시도하는 로직을 구현해야 하기 때문에 별도의 Helper 유틸리티와 같은 Facade를 만들었다. Optimistic Lock에서 Version을 확인하는데, 만약 DB 변경 트랜잭션을 만들고자 하는 현재 쓰레드가 이전 Version을 가지고 있다면 트랜잭션을 보내지 못하고 1ms 동안 기다려야 한다.

// OptimisticLockStockFacade.javapackage com.example.stock.facade;import com.example.stock.service.OptimisticLockStockService;
import org.springframework.stereotype.Service;@Service
public class OptimisticLockStockFacade {private final OptimisticLockStockService optimisticLockStockService;public OptimisticLockStockFacade(OptimisticLockStockService optimisticLockStockService) {
        this.optimisticLockStockService = optimisticLockStockService;
    }public void decrease(final Long id, final Long quantity) throws InterruptedException {
        while (true) {
            try {
                optimisticLockStockService.decrease(id, quantity);
                break;
            } catch (Exception e) {
                // retry
                System.out.println("OPTIMISTIC LOCK VERSION CONFLICT !!!");
                System.out.println(e.getMessage());
                Thread.sleep(1);
            }
        }
    }
}

다음 테스트 코드를 구동해보자. 정상적으로 통과할 것이다. 본인의 컴퓨터에서는 36.494초가 소요되었다. Version 충돌이 잦아 성능이 급격하게 떨어진 것으로 보인다.

@DisplayName("optimistic lock을 사용한 재고 감소 - 동시에 1000개 테스트")
    // 충돌이 빈번하게 일어나지 않을 것이라면 Optimistic Lock을 사용한다.
    @Test
    void OPTIMISTIC_LOCK을_사용한_재고_감소() throws InterruptedException {
        // given// when
        IntStream.range(0, threadCount).forEach(e -> executorService.submit(() -> {
                    try {
                        stockOptimisticLockFacade.decrease(productId, quantity);
                    } catch (final InterruptedException ex) {
                        throw new RuntimeException(ex);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
        ));countDownLatch.await();// then
        final Long afterQuantity = stockRepository.getByProductId(productId).getQuantity();
        System.out.println("### OPTIMISTIC LOCK 동시성 처리 이후 수량 ###" + afterQuantity);
        assertThat(afterQuantity).isZero();
    }

테스트 5. Named Lock의 사용

Named Lock은 이름을 가진 Metadata에 대한 Lock인데, 이름을 가진 Lock을 획득하여 해제할 때까지 다른 세션은 해당 Lock을 획득할 수 없다. Transaction이 종료될 때 Lock이 자동으로 해제되지 않으므로 별도의 명령어를 사용하거나 선점시간이 종료되어 수동 해제처리를 해줘야 한다는 주의점이 있다.

우리는 MySQL의 Native Named Lock을 사용한다. MySQL에서 GET_LOCK과 RELEASE_LOCK 으로 분산 락(distributed lock)을 구현할 수 있다. (참고) 주의할 점은, Named Lock을 활용할 때 데이터소스를 분리하지않고 하나로 사용하게되면 커넥션풀이 부족해지는 현상을 겪을 수 있어서 락을 사용하지 않는 다른 서비스까지 영향을 끼칠 수 있다는 것이다.

Named Lock을 활용하면 분산 락을 구현할 수 있고 Pessmistic Lock은 타임아웃을 구현하기 쉽지만 Named Lock은 타임아웃을 구현하기 쉽다. 그리고 데이터 정합성을 받춰야 하는 경우에도 Named Lock이 좋다. 하지만 트랜잭션 종료 시에 Lock 해제와 세션 관리 (데이터 소스 분리 시) 관리가 수동으로 진행되어야 하고 일일이 수동으로 해야 한다는 불편한 점이 있어 실무 구현에서는 좀 빡세다.

참고로, Pessmistic Lock은 column/row 단계에서 Lock을 걸지만, Named Lock은 metadata 단위에 lock을 건다. 또한, Named Lock에서는 Thread가 아니라 Session이라고 부른다.

다음 자료를 통해서 더욱 이해해보도록 하자. MySQL에서 사용하는 Lock 이해

// NamedLockRepository.javapublic interface NamedLockRepository extends JpaRepository<Stock, Long> {@Query(value = "select get_lock(:key, 1000)", nativeQuery = true)
    void getLock(String key);@Query(value = "select release_lock(:key)", nativeQuery = true)
    void releaseLock(String key);
}------------------------------------------------------------------// NamedLockService.javapackage com.example.stock.service;import com.example.stock.domain.Stock;
import com.example.stock.repository.StockRepository;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;@Service
public class NamedLockStockService implements StockBusinessInterface{private final StockRepository stockRepository;public NamedLockStockService(final StockRepository stockRepository) {
        this.stockRepository = stockRepository;
    }@Transactional(propagation = Propagation.REQUIRES_NEW)
    public synchronized void decrease(final Long id, final Long quantity) {
        final Stock stock = stockRepository.getByProductId(id);
        stock.decrease(quantity);
        stockRepository.saveAndFlush(stock);
    }
}-------------------------------------------------------------------// NamedLockStockFacade.javapackage com.example.stock.facade;import com.example.stock.repository.NamedLockRepository;
import com.example.stock.service.NamedLockStockService;
import org.springframework.stereotype.Component;@Component
public class NamedLockStockFacade {private NamedLockRepository namedLockRepository;private NamedLockStockService namedLockStockService;public NamedLockStockFacade(final NamedLockRepository namedLockRepository, final NamedLockStockService namedLockStockService) {
        this.namedLockRepository = namedLockRepository;
        this.namedLockStockService = namedLockStockService;
    }public void decrease(Long id, Long quantity) {
        try {
            namedLockRepository.getLock(id.toString());
            namedLockStockService.decrease(id, quantity);
        } finally {
            namedLockRepository.releaseLock(id.toString());
        }
    }
}

테스트 코드는 다음과 같다. 본인의 개발 환경에서는 테스트 통과까지 약 21.857초 소요되었다.

@DisplayName("named lock 을 사용한 재고 감소 - 동시에 1000개 테스트 | 21.857s 소요")
    // 데이터 소스를 분리하지 않고 하나로 사용할 경우 커넥션 풀이 부족해질 수 있으므로 분리하는 것을 추천한다.
    @Test
    void NAMED_LOCK을_사용한_재고_감소() throws InterruptedException {
        // given        // when
        IntStream.range(0, threadCount).forEach(e -> executorService.submit(() -> {
                    try {
                        namedLockStockFacade.decrease(productId, quantity);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
        ));countDownLatch.await();        // then
        final Long afterQuantity = stockRepository.getByProductId(productId).getQuantity();
        System.out.println("### NAMED LOCK 동시성 처리 이후 수량 ###" + afterQuantity);
        assertThat(afterQuantity).isZero();
    }

테스트 6. 분산 락 기반의 Lettuce 사용

Redis에 기반한 Lettuce 방식은 setnx (set when not exists) 명령어를 사용하여 분산락을 구현하는 방식이다. Lettuce에서는 spin lock 방식을 구현하여야 하는데, 이는 lock을 해제할 수 있는지 일정 주기에 따라 확인하는 방법이다.

Named Lock과 달리 Redis를 사용하면 트랜잭션에 따라 대응되는 현재 트랜잭션 풀 세션 관리를 하지 않아도 되므로 구현이 편리하다. 앞서 말했듯 Spin Lock 방식이므로 Sleep Time이 적을 수록 Redis에 부하를 줄 수 있어서 thread busy waiting의 요청 간의 시간을 적절히 주어야 한다.

장점으로는 Lettuce 는 Spring Data Redis에서 기존 인터페이스를 제공하기 때문에 러닝 커브가 빠르다. 단점으로는 반드시 수동으로 Lock을 unlock 해주어야 한다.

우리는 분산 락에서 Redis를 사용할 것이기 때문에, 다음과 같이 RedisRepository를 만들어주도록 한다.

// RedisRepository.javapackage com.example.stock.repository;import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;import java.time.Duration;@Component
public class RedisRepository {private final RedisTemplate<String, String> redisTemplate;public RedisRepository(final RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }public Boolean lock(final Long key) {
        String generatedKey = generateKey(key);
        return redisTemplate
                .opsForValue()
                .setIfAbsent(generatedKey, "lock", Duration.ofMillis(3_000));
    }public Boolean unlock(final Long key) {
        String generatedKey = generateKey(key);
        return redisTemplate.delete(generatedKey);
    }public String generateKey(final Long key) {
        return key.toString();
    }
}

이제 Facade 객체를 만들어본다. 이 때 1번 Test처럼 @Transactional 방법을 구현한 기존 서비스를 사용할 것이다.

// LettuceLockStockFacade.javapackage com.example.stock.facade;import com.example.stock.repository.RedisRepository;
import com.example.stock.service.StockNonSynchronizedService;
import com.example.stock.service.StockService;
import org.springframework.stereotype.Component;@Component
public class LettuceLockStockFacade {private final RedisRepository redisRepository;private final StockNonSynchronizedService stockService;public LettuceLockStockFacade(final RedisRepository redisRepository, final StockNonSynchronizedService stockService) {
        this.redisRepository = redisRepository;
        this.stockService = stockService;
    }public void decrease(final Long productId, final Long quantity) throws InterruptedException {
        while (!redisRepository.lock(productId)) {
            Thread.sleep(100); // 부하를 줄여줘본다.
        }try {
            stockService.decrease(productId, quantity);
        } finally {
            redisRepository.unlock(productId);
        }
    }
}

다음은 테스트 코드이다. 본인의 테스트 환경에서는 49.581초 소요되었다.

// StockServiceTest.java@DisplayName("redis lettuce lock 을 사용한 재고 감소")
    // Redis를 사용하면 트랜잭션에 따라 대응되는 현재 트랜잭션 풀 세션 관리를 하지 않아도 되므로 구현이 편리하다.
    // Spin Lock 방식이므로 부하를 줄 수 있어서 thread busy waiting을 통하여 요청 간의 시간을 주어야 한다.
    @Test
    void LETTUCE_LOCK을_사용한_재고_감소() throws InterruptedException {
        // given        // when
        IntStream.range(0, threadCount).forEach(e -> executorService.submit(() -> {
                    try {
                        lettuceLockStockFacade.decrease(productId, quantity);
                    } catch (InterruptedException ex) {
                        throw new RuntimeException(ex);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
        ));countDownLatch.await();        // then
        final Long afterQuantity = stockRepository.getByProductId(productId).getQuantity();
        System.out.println("### LETTUCE LOCK 동시성 처리 이후 수량 ###" + afterQuantity);
        assertThat(afterQuantity).isZero();
    }

테스트 7. 분산 락 기반의 Redisson의 사용

일단 Redisson이 왜 등장했는지에 대한 배경을 이해해보자.

1. 스핀락은 계속해서 Lock 을 획득하기 위해 순회하기 때문에 만약 Lock 을 획득한 스레드나 프로세스가 Lock 을 정상적으로 해제해주지 못한다면 현재 스레드는 계속해서 락을 획득하려 시도하느라 어플리케이션이 중지될 것입니다.

2. 대표적으로 순회 횟수를 5회로 제한한다거나, 아니면 시간으로 제한한다거나를 택할 수 있을 겁니다.

3. setnx 메소드는 만약 키가 존재하지 않는다면 설정하게 되는 것이므로 Redis 에 계속해서 LockKeyName 이 존재하는지 확인해야만 합니다. 따라서 순회하는 동안 계속해서 Redis 에 요청을 보내게 되는 것이므로 스레드 혹은 프로세스가 많다면 Redis 에 부하가 가게 될 것입니다.

4. Lettuce 에서는 Lock 에 대한 기능을 별도로 제공하지 않고, 기존 key-value 를 Setting 하는 방법과 동일하게 사용합니다.

5. 하지만 Redisson 에서는 RLock 이라는 클래스를 따로 제공합니다.

출처: https://devroach.tistory.com/83

Redisson 방식은 Pub-sub 기반으로 Distributed Lock을 구현한다. 채널을 하나 만들고 락을 점유하고 있는 쓰레드가 락을 받으려는 쓰레드에게 점유 해제를 공지한다. 따라서, 별도의 retry 로직이 필요없다.

Redis에서 채널을 사용해보고 싶으시다면 redis-cli를 열어 다음과 같이 따라해보자.

(Session 1) $ docker exec -it 6c7c0a47dd34 redis-cli
(Session 2) $ docker exec -it 6c7c0a47dd34 redis-cli
​
(Session 1) $ subscribe ch1
// Reading messages... (press Ctrl-C to quit)
// 1) "subscribe"
// 2) "ch1"
// 3) (integer) 1
​
(Session 2) $ publish ch1 hello
// (integer) 1
​
(Session 1) $
// 1) "message"
// 2) "ch1"
// 3) "hello"

Redisson은 Lettuce와 달리 별도의 인터페이스이기 때문에 Gradle 의존 패키지 설치 및 별도 Facade 작성이 필요하다. leaseTime을 잘못 잡으면 작업 도중 Lock이 해제될 수도 있으니 주의하도록 한다. 이를 IllegalMonitorStateException 이라고 부른다.

1. Lock 을 해제하는 과정 중 정상적으로 Lock 이 해제가 되지 않는다면 문제가 발생할 수 있는데요. 그래서 Redisson 에서는 LockExpire 를 설정할 수 있도록 해줍니다. 그래서 Redison 의 tryLock Method 에서는 leaseTime 을 설정할 수 있습니다.
​
2. Lock 경과시간 만료후 Lock 에 접근하게 될 수도 있습니다.
만약 A 프로세스가 Lock 을 취득한 후 leaseTime 을 1초로 설정했다고 해봅시다.
근데 A 프로세스의 작업이 2초가 걸리는 작업이였다면 이미 Lock 은 leaseTime 이 경과하여 도중에 해제가 되었을 테고, A 프로세스는 Lock 에 대해서 Monitor 상태가 아닌데 Lock 을 해제하려고 할 것 입니다.
따라서 IllegalMonitorStateException 이 발생하게 됩니다.

출처: https://devroach.tistory.com/83

즉, Lock 획득이 실패하고 재시도가 반드시 필요하지 않은 경우에는 Lettuce를 사용하고, 재시도가 반드시 필요한 경우에는 Redisson을 활용하도록 하자.

이미 RedisRepository는 만들었기 때문에 Facade를 구현하도록 하자. 서비스 객체는 1번 방법에서 도입했던 트랜잭션 기반 방식을 이용해보겠다.

// RedissonLockStockFacade.javapackage com.example.stock.facade;import com.example.stock.service.StockNonSynchronizedService;
import com.example.stock.service.StockService;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;import java.util.concurrent.TimeUnit;@Component
public class RedissonLockStockFacade {private final RedissonClient redissonClient;private final StockNonSynchronizedService stockService;public RedissonLockStockFacade(final RedissonClient redissonClient, final StockNonSynchronizedService stockService) {
        this.redissonClient = redissonClient;
        this.stockService = stockService;
    }public void decrease(final Long productId, final Long quantity) throws InterruptedException {
        final RLock lock = redissonClient.getLock(productId.toString());try {
            boolean isAvailable = lock.tryLock(10, 1, TimeUnit.SECONDS);if (!isAvailable) {
                System.out.println("redisson getLock timeout");
                return;
            }stockService.decrease(productId, quantity);} finally {
            // unlock the lock object
            lock.unlock();
        }
    }
}

다음 테스트를 실행하면 통과할 것이다. 본인의 컴퓨터 환경으로는 약 17.23초가 소요되었다.

// StockServiceTest.java    @DisplayName("redis reddison lock 을 사용한 재고 감소 - 동시에 1000개 테스트 | 17.23s 소요")
    @Test
    void REDISSON_LOCK을_사용한_재고_감소() throws InterruptedException {
        // given// when
        IntStream.range(0, threadCount).forEach(e -> executorService.submit(() -> {
                    try {
                        redissonLockStockFacade.decrease(productId, quantity);
                    } catch (InterruptedException ex) {
                        throw new RuntimeException(ex);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
        ));countDownLatch.await();// then
        final Long afterQuantity = stockRepository.getByProductId(productId).getQuantity();
        System.out.println("### REDDISON LOCK 동시성 처리 이후 수량 ###" + afterQuantity);
        assertThat(afterQuantity).isZero();
    }

Q&A

Pessimistic Lock vs Optimistic Lock

충돌이 적은 경우 optimistic lock 이 빠르지만, 충돌이 많다면 pessimistic lock 이 더 빠르므로, 경우에 따라 다르다.

Facade? Helper?

Facade는 내부 로직을 캡슐화하는 디자인 패턴. 사실 우리 구현사항에서 Facade에는 락을 얻는 행위만 있으므로 다른 패턴이 더 적합할 수 있지만, 구현이 매우 쉬워서 실무에서 자주 쓰는 편이다.

MySQL? Redis?

이미 MySQL 을 사용하고 있다면 별도의 비용 없이 사용가능하다. 어느 정도의 트래픽까지는 문제 없이 활용이 가능하다. ㅏ하지만 Redis 보다는 성능이 좋지 않다.
만약 현재 활용중인 Redis 가 없다면 별도의 구축비용과 인프라 관리비용이 발생한다. 하지만, MySQL 보다 성능이 좋다.

Version 주입할 시의 어노테이션

import javax.persistence.Version;

더 살펴보기

  1. What is the purpose of await() in CountDownLatch? https://stackoverflow.com/questions/41866691/what-is-the-purpose-of-await-in-countdownlatch
  2. MySQL에서 사용하는 Lock 이해http://web.bluecomtech.com/saltfactory/blog.saltfactory.net/database/introduce-mysql-lock.html

참고 문헌

https://devroach.tistory.com/83
https://github.com/Hyune-c/manage-stock-concurrency

Database
Concurrency
Spring Boot
Java
Spring

--

--

Sigrid Jin

Written by Sigrid Jin

501 Followers

Software Engineer at Sionic AI / Kyung Hee Univ. twitter.com/@sigridjin_eth

Help

Status

About

Careers

Press

Blog

Privacy

Terms

Text to speech

Teams