트러블 슈팅 - 동시 구매 시 재고관리에 대한 고민 2. 동시 구매시 재고관리 접근

문제 상황

특정 일시에 오픈하는 예약 구매 특성상 동시에 주문처리가 많이 일어남

이 때에 실제 재고보다 더 많은 수량의 주문이 성공적으로 처리됨

 

문제 원인

주문 시 DB에 짧은 시간 동안 동시 접근이 일어나 여러 트랜잭션에서 동일한 데이터를 수정하려고 할 때 데이터 불일치 문제 발생

 

비즈니스 로직

public class StockServiceImpl implements StockService {

    // 기타 메소드 생략

    @Override
    @Transactional
    public void decreaseStock(Long productId, int count) {
        log.info("재고 감소 요청 productId: {}, count: {}", productId, count);

        Stock stock = stockRepository.findById(productId)
                .orElseThrow(() -> new StockException(ErrorCode.STOCK_NOT_FOUND));

        int newQuantity = stock.getQuantity() - count;
        if (newQuantity < 0) {
            log.warn("재고 불충분 for productId: {}. 요청: {}, 가능: {}", productId, count, stock.getQuantity());
            throw new StockException(ErrorCode.STOCK_NOT_ENOUGH);
        }

        stock.update(newQuantity);
        log.info("재고 감소 완료 for productId: {}. New quantity: {}", productId, newQuantity);
    }
}

@Entity
@Table(name = "product_stock")
@Getter
@Builder
@AllArgsConstructor
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class Stock {

    @Id
    private Long stockId;

    @Column(name = "quantity", nullable = false)
    private Integer quantity;

    // default_batch_fetch_size: 100 적용
    @OneToOne(fetch = FetchType.LAZY)
    @MapsId
    @JsonIgnore
    @JoinColumn(name = "product_id")
    private Product product;

    public void update(Integer newQuantity) {
        this.quantity = newQuantity;
    }
}

 

테스트

@Slf4j
@SpringBootTest
public class StockServiceTests {

    @Autowired
    private StockService stockService;

    @MockBean
    private StockRepository stockRepository;

    private final Long productId = 1L;
    private final int initQuantity = 100;
    private Stock stock;

    // 테스트 객체 초기화
    @BeforeEach
    void setUp() {
        Product product = RegularProduct.builder()
                .productId(1L)
                .name("상품1")
                .content("설명입니다.")
                .price(1000L)
                .isDeleted(false)
                .build();

        stock = Stock.builder()
                .product(product)
                .quantity(initQuantity)
                .build(); // productId가 1인 상품의 초기 재고는 10개

        Mockito.when(stockRepository.findById(product.getProductId()))
                .thenReturn(Optional.of(stock));
    }
    
    // 기타 테스트 메소드

    @DisplayName("멀티스레드 환경에서 재고 차감 테스트")
    @Test
    void decreaseStockInMultiThreadEnvironment() throws InterruptedException {
        int numberOfThreads = 20;
        int decreaseCountPerThread = 1;

        ExecutorService service = Executors.newFixedThreadPool(numberOfThreads);

        for (int i = 0; i < numberOfThreads; i++) {
            log.info("재고 : {}", stock.getQuantity());
            service.submit(() -> {
                try {
                    stockService.decreaseStock(productId, decreaseCountPerThread);
                } catch (Exception e) {
                    log.error("재고감소 실패", e);
                }
            });
        }

        service.shutdown();
        boolean finished = service.awaitTermination(1, TimeUnit.MINUTES);
        assertTrue(finished);

        // 최종 재고 수량 검증
        assertThat(stock.getQuantity()).isEqualTo(initQuantity - numberOfThreads * decreaseCountPerThread);
    }

결과

 

해결 방안

1. 어플리케이션 단에서의 LOCK

2. JPA에서의 LOCK

  2-1. 트랜잭션 격리 레벨

  2-2. 낙관적 락과 비관적 락

3. 분산 LOCK

 

MSA 환경이지만 LOCK에 대하여 공부가 필요하다고 생각되어 하나씩 적용해보았습니다.

 

1. 어플리케이션 단에서의 LOCK

단일 어플리케이션 단에서 사용할 수 있는 락은 synchronized 와 ReentrantLock이 있습니다.

스프링은 멀티 쓰레드 환경을 제공하므로 별도의 쓰레드가 만들어져서 각각의 영역에서 접근이 가능합니다.

이 경우 synchronized 또는 ReentrantLock으로 락을 걸어 임계영역을 설정하면 한 번에 하나의 쓰레드만 접근 할 수 있게 됩니다.

 

Synchronized

public synchronized void decreaseStock(Long productId, int count) {

    // 재고 감소 로직
    
}

public synchronized void restoreStock(Long productId, int count) {

    // 재고 복구 로직

}

먼저 synchronized의 경우 위와 같이 메소드에다가 적용할 수 있으며 테스트 코드 또한 통과 하였습니다.

 

하지만 메소드 레벨에서 synchronized는 하나의 클래스에서 모든 synchronized가 적힌 메소드에서 동일하게 적용됩니다.

즉, "재고 복구" 또한 "재고 감소" LOCK이 해제될 때까지 기다려야 합니다.

 

"재고 복구"는 단순히 구매 취소 뿐 아니라 반품이나 재고 정정 등의 도메인과도 연관이 있을 수 있기 때문에 주문 시 재고 감소에 대한 락을 걸고자 했던 기대값과 다른 사이드 이펙트를 낳습니다.

 

ReentrantLock

// lock 사용을 위한 의존성 주입
private final ConcurrentHashMap<Long, ReentrantLock> locks = new ConcurrentHashMap<>();

public void decreaseStock(Long productId, int count) {
    ReentrantLock lock = locks.computeIfAbsent(productId, k -> new ReentrantLock());
    boolean isLocked = false;
    log.info("재고 감소 요청 productId: {}, count: {}", productId, count);
    try {
        isLocked = lock.tryLock();

        if (!isLocked) {
            throw new StockException(ErrorCode.STOCK_LOCK_FAILURE);
        }

        Stock stock = stockRepository.findById(productId)
                .orElseThrow(() -> new StockException(ErrorCode.STOCK_NOT_FOUND));

        int newQuantity = stock.getQuantity() - count;
        if (newQuantity < 0) {
            log.warn("재고 불충분 for productId: {}. 요청: {}, 가능: {}", productId, count, stock.getQuantity());
            throw new StockException(ErrorCode.STOCK_NOT_ENOUGH);
        }

        stock.update(newQuantity);
    } finally {
        if (isLocked) {
            lock.unlock();
        }
    }
}

 

다음으로 위와같이 수동으로 락을 설정할 수 있는 장점이 있는 reentrantlock을 적용해 봤습니다.

그런데 테스트 코드를 통과하지 못하는 문제가 있었습니다.

코드를 보면 당연한 결과였습니다. tryLock()은 락을 획득하지 못하는 경우 다른 방법으로 들어가기 위한 메소드라 다른 쓰레드에서 락을 얻지 못해 데드락이 걸린 것입니다.

ReentrantLock 리팩토링 방법 두가지

 public void decreaseStock(Long productId, int count) {
    // 기존 로직
    try {
        isLocked = lock.tryLock(5, TimeUnit.SECONDS);

        if (!isLocked) {
            throw new StockException(ErrorCode.STOCK_LOCK_FAILURE);
        }
        // 기존 로직
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        log.warn("락 획득 중 오류 발생");
    } finally {
        // 기존 로직
    }
}

public void decreaseStock(Long productId, int count) {
    // 기존 로직
    try {
        lock.lock();
        //기존 로직
    } finally {
        lock.unlock();
    }
}

 

trylock을 사용하려면 락을 얻지 못한 경우 if문 안에서 다른 로직을 타게끔 해야 합니다.

혹은 리팩토링 첫번째 방법과 같이 오버라이딩 된 메소드로 시간을 정해주고 락을 걸어줄 수 있습니다. 위 코드의 경우 5초의 대기시간을 두고 락을 획득하여 정상적으로 테스트 코드를 통과하였습니다.

 

또는 두번째처럼 lock()메소드를 대신 사용하여 락이 풀릴때까지 대기하는 방법도 있습니다.

이외에도 ReentrantLock은 아래와 같은 기능을 사용할 수 있습니다.

• 공정성 설정 : 가장 오래 기다린 스레드가 락을 얻을 수 있음
• 재진입 가능 : LOCK을 소유한 상태에서 또 락을 얻을 수 있음

해당 방법의 한계

어플리케이션 단에서 LOCK은 하나의 어플리케이션에서는 동시성을 해결 할 수 있는 문제입니다.

그렇지만 MSA 환경에서는 여러 서비스가 같은 데이터에 접근하거나 변경을 시도할 수 있기 때문에 동시성을 해결하기 어렵습니다.

 

2. JPA에서의 LOCK

어플리케이션 단에서의 LOCK은 DB에 접근하기 전 메소드 영역에서 권한을 획득하는 과정이었다면, 이번에는 JPA를 통한 LOCK을 이용하여 DB에서 접근을 제한하여 처리하는 방법입니다.

 

만약 LOCK이 전혀 없다면 아래와 같은 상황이 발생합니다.

트랜잭션1에서 커밋되기 직전의 값을 트랜잭션2에서 읽고 업데이트를 해버린 이후에는 10000 -> 9995로 줄어든 처음 트랜잭션1이 분실되는 Lost update가 발생합니다.

 

2-1. 트랜잭션 격리 레벨

이러한 문제를 트랜잭션 isolation level에서 해결하기 위해서는 Read committed, Repeatable read, Serializable 중 하나의 선택을 고려해 볼 수 있습니다.

Lost update는 Non-repeatable read와 비슷한 맥락으로 이해할 수가 있습니다. 한 트랜잭션 내에서 같은 데이터를 두 번 이상 읽을 때 다른 트랜잭션에 의해 데이터가 변경되어 처음 읽었던 값과 다른 값을 읽으니 이를 업데이트 한다는 것만 빼면 과정이 같습니다.

 

따라서 Non-reapeatable read를 해결하는 Repeatable read와 Serializable를 생각해 볼 수 있습니다.

먼저 Repeatable read는 하나의 트랜잭션에서 데이터를 두 번 읽는 도중 다른 트랜잭션에서 내용을 추가해서 커밋한 경우 다른 결과를 보여주는 Phantom read의 문제를 일으킬 가능성이 있습니다. MySQL에서는 기본값이 Repeatable read이고, 독특한 동작방식 때문에 Phantom read 문제가 발생하지 않는다고 합니다.

 

다음으로 Serializable은 트랜잭션 사이의 모든 간섭을 방지하는 Locking 메커니즘을 제공하여 트랜잭션이 사용중인 모든 행을 잠궈버립니다. 따라서 발생할 수 있는 문제는 해결이 가능하지만 데이터베이스의 동시성을 크게 제한할 수 있습니다.

또한, 데드락의 문제가 발생할 여지가 있습니다. 같은 데이터에 대해 하나는 select * from for update로 x락이 걸리고, serializable 격리 수준에 의해 일반 select시 s락이 걸린다면 동시에 걸릴 수 없기 때문에 무한정 서로간의 해제를 대기할 수 있습니다.

 

따라서 트랜잭션 격리 레벨만 가지고는 동시성 문제를 해결하기 어렵습니다.

 

2-2. 낙관적 락과 비관적 락

JPA에서는 이러한 일관성과 동시성 문제를 해결하기 위해 @Lock 옵션을 제공합니다.

 

낙관적 락(Optimisstic Lock)

낙관적 락은 데이터를 읽을 때 충돌이 발생하지 않을 것이라고 '낙관'하고 사용합니다.

데이터가 변경되면 Version 등을 통해 변경에 대한 정보를 별도로 저장하고, 데이터 UPDATE 시 이를 확인하여 충돌을 확인하는 방법입니다. (version 활용 가능 타입 : long, int, short, timestamp)

@Entity
public class Stock {

    @Id
    private Long stockId;

    @Column(name = "quantity", nullable = false)
    private Integer quantity;

    // default_batch_fetch_size: 100 적용
    @OneToOne(fetch = FetchType.LAZY)
    @MapsId
    @JsonIgnore
    @JoinColumn(name = "product_id")
    private Product product;

    @Version
    private Integer version;
    
    public void update(Integer newQuantity) {
        this.quantity = newQuantity;
    }
}

public interface StockRepository extends JpaRepository<Stock, Long> {

    @Lock(LockModeType.OPTIMISTIC) // 낙관적 락 우선 적용
    Optional<Stock> findById(Long id);
}

 

이렇게 엔티티에 @Version 필드를 만들어 버전을 적용하고, 엔티티가 변경되면 버전이 +1 씩 자동 업데이트가 됩니다.

Repository에 아래 두가지 방식으로 어노테이션을 달아 버전 정보를 적용할 수 있습니다.

@Lock(LockModeType.OPTIMISTIC)	엔티티 변경될 때마다 버전번호를 증가
@Lock(LockModeType.OPTIMISTIC_FORCE_INCREMENT)	엔티티 변경 + 조회 때 모두 버전 번호를 증가

 

만약 동시에 stock 변경을 시도하게 되는 경우라면, 나중에 UPDATE를 시도한 트랜잭션이 version 정보가 불일치하여 롤백하게 됩니다.

이 방식은 실제 DB에 락을 거는 것이 아니라 락 충돌시 어플리케이션 레벨에서 OptimisticLockException를 던지기 때문에 직접 개발자가 예외처리를 해줘야 합니다.

 

충돌이 발생하지 않을 것이라고 가정하기 때문에 현재는 충돌에 대한 문제를 해결하고 있으니 적합하지 않습니다.

 

비관적 락(Pessimistic Lock)

비관적 락은 데이터를 읽을 때 충돌이 발생할 것이라고 가정할 때 사용합니다.

실제 DB에 존재하는 Lock이고, 만약 Lock이 걸려있을 경우 풀린 다음에만 접근이 가능합니다.

 

JPA에서 제공하는 비관적 락 옵션은 다음과 같습니다.

@Lock(LockModeType.PESSIMISTIC_READ)	읽기에 대한 Lock 적용
@Lock(LockModeType.PESSIMISTIC_WRITE)	쓰기에 대한 Lock 적용
@Lock(LockModeType.PESSIMISTIC_FORCE_INCREMENT)	버전 정보 강제 증가

 

먼저, PESSIMISTIC_READ는 읽기에서 일관성을 유지해야 하는 작업에 적합합니다.

예를 들어, 계좌 잔액이 얼마 이상일 때 대출이 가능한 상품이라면 계좌 잔액을 읽는 동안 다른 트랜잭션에 의해 잔액이 변경되지 않아야 하는 경우에 사용할 수 있습니다.

 

다음으로 PESSIMISTIC_WRITE는 현재 상황과 같이 동시에 같은 상품에 대한 구매가 일어날 때 재고 수량을 정확하게 관리하기 위해 사용할 수 있습니다.

현재 변경하고자 하는 트랜잭션에서 이 Lock이 걸리면 다른 트랜잭션이 재고를 읽거나 쓰는 것을 방지할 수 있기 때문에 현재 트랜잭션에서 재고 수량에 대해 일관성과 정확성을 보장할 수 있습니다.

 

마지막으로 PESSIMISTIC_FORCE_INCREMENT는 낙관적 락에서 version을 증가시키는 것처럼 트랜잭션에서 엔티티의 변경이 일어난 경우 버전을 +1 증가시켜서 변경이 있음을 별도로 표현할 수 있습니다.

 

비관적 락 테스트 실패

세가지 중 VERSION을 별도로 관리하지는 않으면서, 현 상황에 맞는 PESSIMISTIC_WRITE를 적용후 테스트 코드를 작동했는데 실패했습니다.

@SpringBootTest
class StockServiceIntegrationTest {

    @Autowired
    private ProductService productService;

    @Autowired
    private StockService stockService;

    @Autowired
    private StockRepository stockRepository;

    @Autowired
    private ProductRepository productRepository;

    @Test
    @Transactional
    void testPessimisticLocking() throws InterruptedException {
        int initQuantity = 10000;

        // 상품과 초기 재고 생성
        ProductCreate productCreate = ProductCreate.builder()
                .name("상품1")
                .content("설명입니다.")
                .price(1000L)
                .quantity(initQuantity)
                .build();

        Long productId = productService.createProductAndStock(productCreate);
        
        // 동시에 재고 감소
        int numberOfThreads = 10;
        int decreaseQuantity = 1;
        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);

        for (int i = 0; i < numberOfThreads; i++) {
            executorService.submit(() -> {
                stockService.decreaseStock(productId, decreaseQuantity);
            });
        }

        executorService.shutdown();

        // 재고 확인
        Stock stock = stockRepository.findById(productId).orElseThrow();
        assertThat(stock.getQuantity()).isEqualTo(initQuantity - numberOfThreads * decreaseQuantity);
    }
}

비관적 락을 적용했는데 낙관적 락에 대한 오류를 던지고 있었습니다.(ObjectOptimisticLockingFailureException)

코드에서 @Version 어노테이션도 찾아보고 낙관적 락이 적용이 된 부분이 있는지 찾아봤지만 보이지 않았습니다.

 

ObjectOptimisticLockingFailureException 오류는 낙관적 락 뿐 아니라 트랜잭션 내에서 다른 트랜잭션을 실행하는 경우에도 발생할 수 있다고 합니다.

IDE에서도 다른 트랜잭션에서의 변화로 인해 ObjectOptimisticLockingFailureException가 떴다고 알려줍니다.

즉, test라는 트랜잭션 안에서 product를 생성하는 트랜잭션이 실행되고 이 메소드안에서 stock을 저장하는 트랜잭션이 또 실행되었는데, 아직 커밋되지 않은 stock에 대해 접근하려고 하니 발생하는 문제입니다.

 

해결 시도1

product와 stock은 OneToOne으로 연결되어 있어서 product 엔티티가 저장이 되어야 stock을 저장할 수 있습니다.

그래서 위 테스트코드에서 EntityManager를 추가하여 product가 저장되면 em.flush(); 를 해봤습니다.

이렇게 하니 executorService 비동기 로직 내부에서 재고 감소 로직을 호출하긴 하지만 stock이 줄지 않는 문제가 발생했습니다.

테스트 코드가 하나의 트랜잭션으로 묶여 있어서 커밋되지 않아 발생한 문제였습니다.

 

해결 시도2

이 문제를 명확히 하기 위해 테스트 코드를 단순화 해봤습니다. product를 저장하고 Stock이 불러오는지 부터 확인을 했습니다.

이를 위해서 테스트코드에서 트랜잭션 어노테이션을 지웠습니다.

@Test
void testCreateProductAndDecreaseStock() throws InterruptedException {
    int initQuantity = 10000;
    // 상품 생성 로직
    ProductCreate productCreate = ProductCreate.builder()
            .name("상품1")
            .content("설명입니다.")
            .price(1000L)
            .quantity(initQuantity)
            .build();

    Long productId = productService.createProductAndStock(productCreate);
    Stock stock = stockRepository.findById(productId)
            .orElseThrow(() -> new StockException(ErrorCode.STOCK_NOT_FOUND));
            
    assertThat(stock.getStockId()).isEqualTo(productId);
}

 

org.hibernate.exception.GenericJDBCException: JDBC exception executing SQL [select s1_0.product_id,s1_0.quantity from product_stock s1_0 where s1_0.product_id=? for update] [Cannot execute statement in a READ ONLY transaction.] [n/a]

 

단순히 stock이 없으면 예외를 던질 것이라고 생각했는데, 실행해보니 Cannot execute statement in a READ ONLY transaction 문제가 있었습니다.

꽤 많은 검색을 헀지만 마땅한 정보를 찾을수가 없었는데, 스프링 데이터 JPA 공식문서를 확인해보니 읽기작업이 기본적으로 @Transactional(readOnly = true)라는 내용을 찾을 수 있었습니다.

@Lock(LockModeType.PESSIMISTIC_WRITE)
Optional<Stock> findById(Long id);

즉, 테스트코드에서 repository를 직접 불러오면서 발생한 문제인데 readOnly에 쓰기 잠금인 비관적 락을 적용해 발생한 문제였습니다.

따라서, 직접 불러오지 않고 조회용 StockQueryService에서 메소드에 트랜잭션을 걸어 가져오는 방법으로 변경하였습니다.

@Service
@RequiredArgsConstructor
public class StockQueryServiceImpl implements StockQueryService {

    private final StockRepository stockRepository;

    /**
     * productId로 Stock 객체를 리턴하는 메소드
     */
    @Override
    @Transactional
    public Stock readStock(Long productId) {
        return stockRepository.findById(productId)
                .orElseThrow(() -> new StockException(ErrorCode.STOCK_NOT_FOUND));
    }
}

@Test
void testCreateProductAndDecreaseStock() throws InterruptedException {
    int initQuantity = 10000;
    // 상품 생성 로직
    ProductCreate productCreate = ProductCreate.builder()
            .name("상품1")
            .content("설명입니다.")
            .price(1000L)
            .quantity(initQuantity)
            .build();

    Long productId = productService.createProductAndStock(productCreate);
    Stock stock = stockQueryService.readStock(productId);
    
    assertThat(stock.getStockId()).isEqualTo(productId);
}

 

이렇게 변경 한 후 정상적으로 Stock 객체를 가져오는 것을 확인했습니다.

@SpringBootTest
class StockServiceIntegrationTest {

    @Autowired
    private ProductService productService;

    @Autowired
    private StockService stockService;

    @Autowired
    private StockQueryService stockQueryService;

    @Autowired
    private StockRepository stockRepository;

    @Test
    void testCreateProductAndDecreaseStock() throws InterruptedException {
        int initQuantity = 10000;
        // 상품 생성 로직
        ProductCreate productCreate = ProductCreate.builder()
                .name("상품1")
                .content("설명입니다.")
                .price(1000L)
                .quantity(initQuantity)
                .build();

        Long productId = productService.createProductAndStock(productCreate);

        int numberOfThreads = 10;
        int decreaseQuantity = 1;

        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
        AtomicInteger failCount = new AtomicInteger();
        CountDownLatch countDownLatch = new CountDownLatch(numberOfThreads);

        for (int i = 0; i < numberOfThreads; i++) {
            executorService.execute(() -> {
                try {
                    stockService.decreaseStock(productId, decreaseQuantity);
                    System.out.println("성공");
                } catch (PessimisticLockingFailureException iae) {
                    System.out.println("iae.getMessage() = " + iae.getMessage());
                    failCount.getAndIncrement();
                } catch (Exception e) {
                    System.out.println("비관적 락 발생");
                    System.out.println("e.getCause() = " + e.getCause());
                    System.out.println("e = " + e);
                    failCount.getAndIncrement();
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();

        // 재고 확인
        Stock stock = stockQueryService.readStock(productId);
        assertThat(stock.getQuantity()).isEqualTo(initQuantity - (numberOfThreads * decreaseQuantity));
    }
}

 

그리고 다시 코드를 작성하여 다소 지저분하지만, 위와같은 코드를 실행시켰고 정상적으로 통과함을 확인할 수 있었습니다.

 

해결 시도3

@SpringBootTest
class StockServiceIntegrationTest2 {

    @Autowired
    private ProductService productService;

    @Autowired
    private StockService stockService;

    @Autowired
    private StockRepository stockRepository;

    private final int INIT_QUANTITY = 10000;

    @Test
    @Transactional
    @Commit
    void testCreateProductAndCommit() {
        // 상품 생성 로직
        ProductCreate productCreate = ProductCreate.builder()
                .name("상품1")
                .content("설명입니다.")
                .price(1000L)
                .quantity(INIT_QUANTITY)
                .build();
        productService.createProductAndStock(productCreate);
    }

    @Test
    @Transactional
    void testPessimisticLocking() throws InterruptedException {
        // 동시에 재고를 감소시키는 시뮬레이션
        int numberOfThreads = 10;
        int decreaseQuantity = 1;
        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
        AtomicInteger failCount = new AtomicInteger();
        CountDownLatch countDownLatch = new CountDownLatch(numberOfThreads);

        for (int i = 0; i < numberOfThreads; i++) {
            executorService.execute(() -> {
                try {
                    stockService.decreaseStock(1L, decreaseQuantity);
                    System.out.println("성공");
                } catch (PessimisticLockingFailureException iae) {
                    System.out.println("iae.getMessage() = " + iae.getMessage());
                    failCount.getAndIncrement();
                } catch (Exception e) {
                    System.out.println("비관적 락 발생");
                    System.out.println("e.getCause() = " + e.getCause());
                    System.out.println("e = " + e);
                    failCount.getAndIncrement();
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();

        // 재고 확인
        Stock stock = stockRepository.findById(1L)
        		.orElseThrow(() -> new StockException(ErrorCode.STOCK_NOT_FOUND));
        assertThat(stock.getQuantity()).isEqualTo(INIT_QUANTITY - (numberOfThreads * decreaseQuantity));
    }
}

트랜잭션별로 분류하는 방법도 생각해봤습니다.

위 메소드에서 product를 @Commit 어노테이션으로 DB에 저장하고, 아래 메소드로 테스트하니 정상적으로 재고 감소가 반영됨을 확인할 수 있었습니다.

 

하지만 이 코드의 문제는 jpa.hibernate.ddl-auto가 create 일 때 먼저 커밋이 된 다음에야 테스트가 가능하므로 독립적으로 테스트를 할 수가 없습니다. ddl-auto를 update로 하면 되지만 이 또한 첫번째 테스트 메소드가 예상하지 못한 영향을 끼칠 수도 있기에 올바른 테스트 방법은 아니라고 생각하여 폐기했습니다.

 

결론

여러가지 테스트를 진행했지만 결국 이 또한 상황에 따라 데드락이 발생할 여지가 있습니다.

또한, 이러한 DB락은 MSA환경에서처럼 다중 서버를 사용하는 경우 엔티티에 대한 Lock을 보장하지 못합니다.

 

3. 분산 락(distributed lock)

분산 락은 MSA와 같은 분산 환경에서 상호 배제를 구현하여 동시성 문제를 다루는 락입니다.

DB에서 락을 거는 것이 아니라 락에 대한 정보를 위임하여 임계 영역(critical section)에 접근할 수 있는지 확인합니다.

이를 통해 분산 환경에서 락을 획득하고 동시성 문제를 해결 할 수 있습니다.

 

사용 이유

MySQL에서 제공하는 분산락을 사용해볼까하다 프로젝트 기한 마감을 약 4주로 잡았기 때문에 빠른 구현을 위해 Redis의 Redisson을 사용하였습니다.

 

Redisson에는 많은 기능과 장점들이 있지만 분산락 기능에만 집중하여 스프링의 AOP와 커스텀 어노테이션을 이용하여 사용했습니다.

이렇게 하면 어노테이션만으로 락을 사용할 수 있어 재사용성을 높일 수 있기 때문입니다.

 

build.gradle

dependencies {
    // 기타 생략
	implementation 'org.springframework.boot:spring-boot-starter-data-redis'
	implementation "org.redisson:redisson-spring-data-27:3.27.0"
}

RedissonConfig

@Configuration
public class RedissonConfig {

    @Value("${spring.redis.host}")
    private String host;

    @Value("${spring.redis.port}")
    private int port;

    private static final String REDISSON_HOST_PREFIX = "redis://";

    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        config.useSingleServer().setAddress(REDISSON_HOST_PREFIX + host + ":" + port);

        return Redisson.create(config);
    }

}

LockConfig

@Getter
@RequiredArgsConstructor
public enum LockConfig {

    TEST_LOCK(1000L, 3000L); // lock 시간 설정

    private final long waitTimeOutMills;
    private final long leaseTimeoutMills;
}

LockAspect

@Component
@Aspect
@RequiredArgsConstructor
@Slf4j
public class LockAspect {

    private final RedissonClient redissonClient;

    @Around("@annotation(distributedLock)")
    public Object executeFunctionWithLock(ProceedingJoinPoint proceedingJoinPoint, DistributedLock distributedLock)
            throws Throwable {
        LockConfig lockConfig = distributedLock.lockConfig();
        String lockName = lockConfig.name();
        String methodName = proceedingJoinPoint.getSignature().getName();

        log.info("{} 에서 lock: {} 을 획득하려 시도", methodName, lockName);
        var lock = redissonClient.getLock(lockName);

        boolean tryLock = lock.tryLock(lockConfig.getWaitTimeOutMills(), lockConfig.getLeaseTimeoutMills(),
                TimeUnit.MILLISECONDS);

        if (!tryLock) {
            log.error("LOCK을 획득할 수 없습니다.");
            throw new IllegalStateException("Unable to acquire lock");
        }

        try {
            log.info("{} 에서 lock: {} 을 획득하려 시도", methodName, lockName);
            return proceedingJoinPoint.proceed();
        } finally {
            lock.unlock();
            log.info("{} 에서 lock: {} 을 해제", methodName, lockName);
        }
    }
}

DistributedLock

/**
 * redisson으로 분산락을 제공하는 어노테이션
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface DistributedLock {

    LockConfig lockConfig();
}

 

실제 사용

@Override
@Transactional
@DistributedLock(lockConfig = LockConfig.TEST_LOCK)  // Lock이 필요한 메소드에 어노테이션
public void decreaseStock(Long productId, int count) {
	// 메소드 생략
}

 

redisson을 이용하여 모든 테스트코드를 돌려보니 문제없이 잘 동작하는 것을 확인할 수 있었습니다.

이미 구현되어있는 것을 가져다 쓴 것이라 간단하게 구현했지만, redisson에 대해 더 공부해보면 좋을 것 같다는 생각이 들었습니다.

 

결론

이렇게 어플리케이션 -> 데이터베이스 -> 분산 환경 순으로 하나씩 적용시켜보았습니다.

DB접근과 동시성을 고민하다보니 자연스레 해결해야하는 규모가 점점 커지게 되었고, 지금과 같은 rest api형태가 아니라 이벤트 발행/구독 형태의 kafka로 MSA를 구현해보는 것도 좋겠다는 것이 숙제로 남았습니다.

 

또한, 구현에 급급해 테스트코드에 대해 거의 적지 못했었는데 블로그로 정리하면서 테스트코드가 왜 중요한지 알게되어 jUnit에 대한 공부가 필요하다는 점, 그리고 차마 결과를 분석하지 못해 블로그 내용엔 적지 못했지만 jMeter를 이용한 성능테스트도 더 공부해보면 좋겠다는 생각이 들었습니다.

 

참고자료

동시성 문제(synchronized, pessimistic lock, optimistic lock)

동시성 이슈 해결방법 (Synchronized, DB Lock, Redis Lock)

Transactionality :: Spring Data JPA

자바 동기화 블록(Java Synchronized Blocks)

https://www.youtube.com/watch?v=bLLarZTrebU

InnoDB는 Repeatable Read에서 어떻게 Phantom Read를 예방할까?

MySQL을 이용한 분산락으로 여러 서버에 걸친 동시성 관리 | 우아한형제들 기술블로그

자바에서 동시성을 해결하는 다양한 방법과 Redis의 분산락

[redis] redisson을 통한 분산 락