[Spring Boot] 동시성 제어 with 비관적 락, Redis 그리고 @Transactional 사용 시 동시성 문제점

1. 개요

현재 개발 중인 것은 모의 주식 사이트로 매수, 매도 주문이 발생한다. 당연히 이러한 주문은 동시성 제어가 되어야하며 여러 방법들을 사용해보면서 성능과 함께 비교해보겠다. 

 

    /**
     * 주문 생성 - 시장가 주문
     */
    @Override
    @Transactional
    public synchronized BuyOrderDto createMarketOrder(BuyOrderRequest buyOrderRequest){
        // 현재 주문 금액 조회
        BigDecimal currentPrice = stockInfoHolder.getCurrentPrice(buyOrderRequest.getSymbol());

        // 계좌 조회 및 금액 차감
        Account account = accountRepository.findByAccountNumber(buyOrderRequest.getAccountNumber());
        account.buyByUSD(currentPrice.multiply(buyOrderRequest.getQuantity()));
        accountRepository.save(account);

        // 주문 생성
        BuyOrder order = BuyOrder.of(buyOrderRequest, stockRepository.getReferenceById(buyOrderRequest.getSymbol()), account, currentPrice);
        BuyOrder saveOrder = buyOrderRepository.save(order);

        // 보유 주식 추가
        holdingStockRepository.findByAccount_AccountNumberAndStockSymbol(buyOrderRequest.getAccountNumber(), buyOrderRequest.getSymbol())
                .ifPresentOrElse(
                        holdingStock -> updateExistingHoldingStock(holdingStock, buyOrderRequest, currentPrice),
                        () -> createNewHoldingStock(account, buyOrderRequest, currentPrice)
                );

        return BuyOrderDto.fromEntity(saveOrder);
    }

 

동시성 제어 테스트를 할 메서드는 위와 같다.

 

1. 매수를 위해 현재 금액을 StockInfoHolder 를 통해 조회한다. StockInfoHolder 매 순간 업데이트 되며 동기화를 위해 Concurrent 패키지를 사용하여 금액을 관리하고 있다. 
2. 실제 계좌를 조회하고 구매하는 만큼 잔액을 차감한다. 
3. 매수 주문을 생성하고 저장한다.
4. 계좌에 보유 주식을 추가한다.
   1. 이미 보유 중이라면 해당 레코드를 조회 후 수량만 추가한다.
   2. 보유 중인 주식이 아니라면 새로운 레코드를 데이터베이스에 추가한다.

 

2. 테스트 코드 및 문제 파악

@SpringBootTest
@ActiveProfiles("test")
@TestPropertySource(locations = "classpath:application-test.yml")
@Slf4j
class BuyOrderServiceConcurrencyTest {

    @Autowired
    private BuyOrderServiceImpl buyOrderService;

    @Autowired
    private AccountRepository accountRepository;

    @Autowired
    private HoldingStockRepository holdingStockRepository;

    @Autowired
    private StockRepository stockRepository;

    @MockBean
    private StockInfoHolder stockInfoHolder;

    private static final String ACCOUNT_NUMBER = "TEST-ACC-001";
    private static final String STOCK_SYMBOL = "BTCUSDT";
    private static final BigDecimal INITIAL_BALANCE = BigDecimal.valueOf(5000);
    private static final BigDecimal STOCK_PRICE = BigDecimal.valueOf(100.00);
    private static final int CONCURRENT_USERS = 30;
    private static final BigDecimal ORDER_QUANTITY = BigDecimal.ONE;

    @BeforeEach
    @Transactional
    void setUp() {
        // 테스트 계좌 생성
        Account testAccount = Account.builder()
                .accountNumber(ACCOUNT_NUMBER)
                .usdBalance(INITIAL_BALANCE)
                .build();
        accountRepository.save(testAccount);

        // StockInfoHolder Mock 설정 (만약 Mock 사용시)
        when(stockInfoHolder.getCurrentPrice(STOCK_SYMBOL))
                .thenReturn(STOCK_PRICE);
    }

    @Test
    void testConcurrentMarketOrders() throws InterruptedException {
        // Given
        int numberOfThreads = CONCURRENT_USERS;
        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
        CountDownLatch latch = new CountDownLatch(numberOfThreads);
        CyclicBarrier barrier = new CyclicBarrier(numberOfThreads);
        List<Future<BuyOrderDto>> futures = new ArrayList<>();

        // When: 여러 스레드가 동시에 주문 생성
        for (int i = 0; i < numberOfThreads; i++) {
            futures.add(executorService.submit(() -> {
                try {
                    barrier.await(); // 모든 스레드가 동시에 시작하도록 대기

                    BuyOrderRequest request = BuyOrderRequest.builder()
                            .accountNumber(ACCOUNT_NUMBER)
                            .symbol(STOCK_SYMBOL)
                            .quantity(ORDER_QUANTITY)
                            .orderType(OrderType.MARKET)
                            .build();

                    return buyOrderService.createMarketOrder(request);
                } catch (Exception e){
                    log.error("Error occurred during order creation", e);
                    return null;
                } finally {
                    latch.countDown();
                }
            }));
        }

        // 모든 스레드가 완료될 때까지 대기
        latch.await();
        executorService.shutdown();

        // Then
        // 1. 모든 주문이 성공적으로 처리되었는지 확인
        List<BuyOrderDto> completedOrders = futures.stream()
                .map(this::getFutureResult)
                .filter(order -> order != null)
                .toList();
        assertThat(completedOrders).hasSize(numberOfThreads);

        // 2. 계좌 잔액이 정확한지 확인
        Account finalAccount = accountRepository.findByAccountNumber(ACCOUNT_NUMBER);
        BigDecimal expectedBalance = INITIAL_BALANCE.subtract(
                STOCK_PRICE.multiply(ORDER_QUANTITY.multiply(BigDecimal.valueOf(numberOfThreads)))
        );
        assertThat(finalAccount.getUsdBalance()).isEqualByComparingTo(expectedBalance);

        // 3. 보유 주식 수량이 정확한지 확인
        HoldingStock holdingStock = holdingStockRepository
                .findFirstByAccount_AccountNumberAndStockSymbol(ACCOUNT_NUMBER, STOCK_SYMBOL)
                .orElseThrow();
        BigDecimal expectedQuantity = ORDER_QUANTITY.multiply(BigDecimal.valueOf(numberOfThreads));
        assertThat(holdingStock.getQuantity()).isEqualByComparingTo(expectedQuantity);
    }

    private BuyOrderDto getFutureResult(Future<BuyOrderDto> future) {
        try {
            return future.get(5, TimeUnit.SECONDS);
        } catch (Exception e) {
            return null;
        }
    }
}

 

우선 5000 USD 가 들어있는 계좌를 생성한다. 그리고 100 USD 가격으로 1 BTCUSDT 매수 주문이 30번 발생할 예정이다. 

모두 시장가 주문이므로 즉시 체결될 예정이고 검증은 세 단계에 걸쳐서 일어난다.

 

1. 모든 주문이 정상적으로 처리되었는 지 확인하기 위해 30개의 return 값이 존재하는 지 확인한다. 

2. 계좌 잔액이 (5000 - 100 * 1 * 30) = 2000 USD 인 지 확인한다.

3. 보유 주식을 조회해 BTCUSDT 가 30개 존재하는 지 확인한다. 

 

2024-11-19T22:41:00.509+09:00  WARN 7964 --- [pool-2-thread-8] o.h.engine.jdbc.spi.SqlExceptionHelper   : SQL Error: 1213, SQLState: 40001
2024-11-19T22:41:00.510+09:00 ERROR 7964 --- [pool-2-thread-8] o.h.engine.jdbc.spi.SqlExceptionHelper   : Deadlock found when trying to get lock; try restarting transaction
2024-11-19T22:41:00.525+09:00 ERROR 7964 --- [pool-2-thread-5] i.s.d.o.s.BuyOrderServiceConcurrencyTest : Error occurred during order creation
org.springframework.dao.CannotAcquireLockException: could not execute statement [Deadlock found when trying to get lock; try restarting transaction] [update accounts set account_number=?,bitcoin_balance=?,krw_balance=?,usd_balance=?,user_id=? where id=?]; SQL [update accounts set account_number=?,bitcoin_balance=?,krw_balance=?,usd_balance=?,user_id=? where id=?]
Caused by: org.hibernate.exception.LockAcquisitionException: could not execute statement [Deadlock found when trying to get lock; try restarting transaction] [update accounts set account_number=?,bitcoin_balance=?,krw_balance=?,usd_balance=?,user_id=? where id=?]

...

at jdk.proxy3/jdk.proxy3.$Proxy203.findFirstByAccount_AccountNumberAndStockSymbol(Unknown Source) ~[na:na]
at com.mock.investment.stock.domain.order.service.BuyOrderServiceImpl.createMarketOrder(BuyOrderServiceImpl.java:62)

 

동시성 제어가 되지 않은 테스트 코드를 실행시키면 당연히 에러가 발생한다. 에러 코드는 위와 같은데, 데드락이 발생하기에 더 이상 실행이 안되는 것이다. 

 

					catch (Exception e) {
                    // 기존 로깅 유지
                    log.error("Error occurred during order creation", e);

                    // 데드락 상세 정보 조회 추가
                    try (Connection conn = dataSource.getConnection();
                         Statement st = conn.createStatement();
                         ResultSet rs = st.executeQuery("SHOW ENGINE INNODB STATUS")) {
                        if (rs.next()) {
                            log.error("InnoDB Status at time of deadlock:\n{}", rs.getString("Status"));
                        }
                    } catch (SQLException ex) {
                        log.error("Failed to get InnoDB status", ex);
                    }
                    return null;
                }

 

데드락 발생 시 위 코드를 통해 실제 InnoDB 의 Status 를 불러오는 작업을 진행했다. 

=====================================
2024-11-19 14:06:04 140444562294528 INNODB MONITOR OUTPUT
=====================================

...

------------------------
LATEST DETECTED DEADLOCK
------------------------
2024-11-19 14:02:17 140444625270528
*** (1) TRANSACTION:
TRANSACTION 44275, ACTIVE 0 sec starting index read
mysql tables in use 1, locked 1
LOCK WAIT 7 lock struct(s), heap size 1128, 3 row lock(s), undo log entries 1
MySQL thread id 552, OS thread handle 140444560180992, query id 44204 172.24.0.1 min9805 updating
update accounts set account_number='TEST-ACC-001',bitcoin_balance=null,krw_balance=null,usd_balance=4700.00,user_id=null where id=2

*** (1) HOLDS THE LOCK(S):
RECORD LOCKS space id 1489 page no 4 n bits 72 index PRIMARY of table `stock`.`accounts` trx id 44275 lock mode S locks rec but not gap
Record lock, heap no 3 PHYSICAL RECORD: n_fields 8; compact format; info bits 0
 0: len 8; hex 8000000000000002; asc         ;;
 1: len 6; hex 00000000ace7; asc       ;;
 2: len 7; hex 010000019121ea; asc      ! ;;
 3: SQL NULL;
 4: SQL NULL;
 5: len 17; hex 800000000000000000000000000012c000; asc                  ;;
 6: SQL NULL;
 7: len 12; hex 544553542d4143432d303031; asc TEST-ACC-001;;


*** (1) WAITING FOR THIS LOCK TO BE GRANTED:
RECORD LOCKS space id 1489 page no 4 n bits 72 index PRIMARY of table `stock`.`accounts` trx id 44275 lock_mode X locks rec but not gap waiting
Record lock, heap no 3 PHYSICAL RECORD: n_fields 8; compact format; info bits 0
 0: len 8; hex 8000000000000002; asc         ;;
 1: len 6; hex 00000000ace7; asc       ;;
 2: len 7; hex 010000019121ea; asc      ! ;;
 3: SQL NULL;
 4: SQL NULL;
 5: len 17; hex 800000000000000000000000000012c000; asc                  ;;
 6: SQL NULL;
 7: len 12; hex 544553542d4143432d303031; asc TEST-ACC-001;;


*** (2) TRANSACTION:
TRANSACTION 44280, ACTIVE 0 sec starting index read
mysql tables in use 1, locked 1
LOCK WAIT 7 lock struct(s), heap size 1128, 3 row lock(s), undo log entries 1
MySQL thread id 557, OS thread handle 140444563351296, query id 44206 172.24.0.1 min9805 updating
update accounts set account_number='TEST-ACC-001',bitcoin_balance=null,krw_balance=null,usd_balance=4700.00,user_id=null where id=2

...

----------------------------
END OF INNODB MONITOR OUTPUT
============================

 

그 결과 위 같은 로그를 받아볼 수 있었고 트랜잭션 44275와 44280이 동일한 accounts 테이블의 id=2 레코드를 대상으로 작업하는 과정에서 데드락이 발생하는 것을 직접 확인할 수 있었다. 두 트랜잭션 모두 이미 조회가 이루어져 S lock 을 보유 중이였고 둘 다 UPDATE 를 위해 X lock 획득을 시도하고 있었다. 하지만 서로의 S lock 으로 인해 둘 다 X lock 을 획득하지 못하고 데드락에 빠져버리는 것이다!! 

 

로그를 통해 정확한 원인을 파악했으니 이를 해결하고 동시성 문제도 해결해보자.

3. 동시성 제어

(1) Synchronized

Java 의 대표적인 동기화 키워드이다. 단순하게 생각해보면 Synchronized 키워드가 붙은 메서드는 한 번에 하나의 스레드만 접근할 수 있기 때문에 동시성 문제가 해결될 수 있을 거라 생각된다.

    @Override
    @Transactional
    public synchronized BuyOrderDto createMarketOrder(BuyOrderRequest buyOrderRequest){ ... }

 

키워드 추가만으로 간단하게 구현할 수 있다. 하지만 여기에서는 고작 3개의 주문만 체결되었다. 

 

주문 자체가 실패하는 경우가 많은 게 이상해서 찾아보니 에러 문구가 "Query did not return a unique result." 이다.

현재 Optional<Entity> 타입으로 받아오는 find 메서드가 있는 데 동시성 문제로 여러 레코드가 한 번에 save 되다보니 Optional 이지만 2개 이상의 결과값이 찾아져서 에러가 발생해 정상적으로 동작하는 것은 오직 3 개의 스레드가 되는 문제였다. 

 

원래 해당 find 결과값은 1개이거나 없는 것이 맞으므로 정상적인 결과이지만 동시성 문제를 좀 더 눈에 보기 위해 JPA 쿼리문을 "findFirst..." 로 변경하였다. 이렇게되면 2개 이상이 결과가 찾아지더라도 한 개만 가져오니 위 같은 에러는 발생하지 않고 실제 남은 금액에서 검증에 실패하게 된다.

@Override
public synchronized BuyOrderDto createMarketOrder(BuyOrderRequest buyOrderRequest){

 

하지만 한 번에 하나의 쓰레드만 실행되는 데 실패하는 게 이상하지 않은가? 그래서 @Transcational 어노테이션을 제거해봤다. 그 결과 아래처럼 모든 검증을 통과하였다. 

 

그렇다면 @Transactional 과 Synchronized 를 함께 쓰면 문제가 발생할 수 있다는 것인데 어떤 문제일까?

 

1) @Transactional 과 Synchronized

@Transactional 어노테이션을 사용할 때에는 AOP 에 의해 proxy 인스턴스가 생성된다. synchronized 메서드에 @Transactional 을 붙이면 아래와 같이 동작한다. 

class BuyOrderServiceProxy {

     BuyOrderServiceImpl buyOrderServiceImpl;

     public void demo() {
          try {
               transaction.start()
               buyOrderServiceImpl.createMarketOrder()
          } catch(Exception e){
               transaction.rollback()
          } finally {
               transaction.commit()
          }
     }
}

class BuyOrderServiceImpl {

     synchronized public BuyOrderDto createMarketOrder() {
           // 매도 주문 처리
     }

}

 

proxy 인스턴스로 메서드가 감싸지는 것을 확인할 수 있다. 하지만 synchronized 키워드는 메서드까지만 적용되고 proxy 인스턴스 전체에 적용되지는 않는다. 다시 말해 트랜잭션이 시작되고, 메서드가 처리 된 다음에 커밋되기 이전 틈이 존재한다는 것이다. 

 

동시에 많은 스레드들이 동작한다면 synchronized 가 걸려있더라도 커밋되기 이전에 값을 읽어 사용하기에 동시성 제어가 안되는 것이다. REPEATABLE READ 를 사용하면서 동시에 동일한 레코드를 업데이트하기 때문에 검증이 안되는 부분은 따로 설명하지 않겠다. 

 

이를 처리하기 위해서는 Transactional 을 제거하거나, Synchronized 메서드 내부에 Transactional 을 필요한 부분에만 적용시키면 된다. 하지만 이는 근본적인 해결책은 아니기 때문에 다음 방법을 살펴보자.

(2) 비관적 락

데이터베이스의 비관적 락은 충돌할 것을 가정에 두고 트랜잭션이 시작될 때 X lock 을 거는 방법이다. 위에서 데드락에 빠지는 이유는 두 개 이상의 트랜잭션이 S lock 을 소유한 상태에서 X lock 을 획득하려하기 때문이다. 처음부터 X lock 을 획득하도록 바꾸게 된다면 한 번에 하나의 트랜잭션만 락을 소유하기에 데드락에 빠지지 않으면서 동시성 제어가 가능하다. 

public interface AccountRepository extends JpaRepository<Account, Long> {
	List<Account> findByUserId(Long userId);

	@Lock(LockModeType.PESSIMISTIC_WRITE)
	@Query("SELECT a FROM Account a WHERE a.accountNumber = :accountNumber")
	Account findByAccountNumberWithLock(String accountNumber);

	Account findByAccountNumber(String accountNumber);
}

 

JPA 를 사용하고 있다면 Lock 은 어노테이션을 통해 가능하다. 아래와 같이 비관적 락을 명시적으로 걸어주면 X lock 을 획득하게 된다.

	@Override
    @Transactional
    public BuyOrderDto createMarketOrder(BuyOrderRequest buyOrderRequest){
        // 현재 주문 금액 조회
        ...
        
        // 계좌 조회 및 금액 차감
        Account account = accountRepository.findByAccountNumberWithLock(buyOrderRequest.getAccountNumber());
        ...
        
        // 주문 생성
        ...
        
        // 보유 주식 추가
		...
    }

 

실제 코드에서는 find 메서드를 withLock 으로 변경만 해주었다. 

 

그 결과 동시성 문제가 해결되는 것을 확인할 수 있다. 이게 가능한 것은 @Transactional 안에서 Lock 을 걸게되면 트랜잭션이 끝날 때까지 Lock 을 보유하기 때문이다. 

 

Account 조회에서만 Lock 을 걸어주게 되면 이후 HoldingStock 에서도 Lock 을 걸어줘야한다고 생각할 수 있다. 하지만 실제로는 Account 에서 Lock 을 걸고 트랜잭션이 끝나야 해제되기 때문에 이후 과정도 Lock 으로 제어되고 있다고 볼 수 있다. 

 

따라서 위의 방법보다는 동시성 제어가 확실히 필요한 부분에 대해서만 트랜잭션을 분리하고 Lock 을 최대한 짧게 유지하는 방법이 권장된다. 

 

Lock 을 통한 병렬성을 유지하고 있기 떄문에 동일한 테스트 코드에서도 Synchronized 에 비해 50% 줄어든 실행 시간을 확인할 수 있다. 

 

(3) Redis 분산락

1) Spin Lock

/**
 * 주문 생성 - 시장가 주문
 */
@Override
@Transactional
public BuyOrderDto createMarketOrder(BuyOrderRequest buyOrderRequest) {
    String lockKey = "LOCK:" + buyOrderRequest.getAccountNumber();
    String lockValue = UUID.randomUUID().toString();

    long startTime = System.currentTimeMillis();
    long timeout = 5000; // 5초 동안 재시도

    while (System.currentTimeMillis() - startTime < timeout) {
        try {
            // 락 획득 시도
            Boolean acquired = redisTemplate.opsForValue()
                    .setIfAbsent(lockKey, lockValue, 1, TimeUnit.SECONDS);

            if (Boolean.TRUE.equals(acquired)) {
                try {
                    // 비즈니스 로직 수행
                    ...

                    return BuyOrderDto.fromEntity(saveOrder);
                } finally {
                    redisTemplate.delete(lockKey);
                }
            }

            // 락 획득 실패시 잠시 대기
            log.debug("Failed to acquire lock. Retrying...");
            Thread.sleep(100);

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Lock acquisition was interrupted", e);
        }
    }

    throw new RuntimeException("Failed to acquire lock after " + timeout + "ms");
}

 

우선 Redis 의 기본 라이브러리는 Lettuce 가 사용된다. 해당 라이브러리를 사용해서 분산락을 구현한다면 락을 획득하기 위해서 SETNX 명령어를 사용해 락 획득 요청을 보낸다. 이를 구현하기 위해서는 스핀락 형태를 사용해야하고 Redis 의 부담을 줄이기 위해서는 중간에 sleep 설정도 필요하다.

 

실제 동시성 문제도 해결 가능하지만 자체적인 타임아웃이나 스핀락 방식을 직접 구현해줘야하는 문제가 있다. 

2) AOP 분산락 

범용성 있게 사용하기 위해 AOP 로 분산락을 만들어 사용해보겠다. 

@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface DistributedLock {
    String key();
    long waitTime() default 5000L;
    long leaseTime() default 2000L;
}

@Slf4j
@Aspect
@Component
@RequiredArgsConstructor
public class DistributedLockAspect {
    private final RedisTemplate<String, String> redisTemplate;
    private final ExpressionParser parser = new SpelExpressionParser();

    @Around("@annotation(distributedLock)")
    public Object lock(ProceedingJoinPoint joinPoint, DistributedLock distributedLock) throws Throwable {
        String lockKey = obtainLockKey(joinPoint, distributedLock);
        String lockValue = UUID.randomUUID().toString();

        long startTime = System.currentTimeMillis();
        long waitTime = distributedLock.waitTime();

        // 설정된 대기 시간 동안 락 획득 시도
        while (System.currentTimeMillis() - startTime < waitTime) {
            Boolean acquired = redisTemplate.opsForValue()
                    .setIfAbsent(lockKey, lockValue, distributedLock.leaseTime(), TimeUnit.MILLISECONDS);

            if (Boolean.TRUE.equals(acquired)) {
                try {
                    log.info("Lock acquired for key: {}", lockKey);
                    return joinPoint.proceed();
                } finally {
                    redisTemplate.delete(lockKey);
                    log.info("Lock released for key: {}", lockKey);
                }
            }

            // 락 획득 실패시 잠시 대기 후 재시도
            try {
                Thread.sleep(100);  // 100ms 대기
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Lock acquisition interrupted", e);
            }
        }

        // 최대 대기 시간 초과
        throw new RuntimeException("Failed to acquire lock after " + waitTime + "ms for key: " + lockKey);
    }

    private String obtainLockKey(ProceedingJoinPoint joinPoint, DistributedLock distributedLock) {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        String[] parameterNames = signature.getParameterNames();
        Object[] args = joinPoint.getArgs();

        EvaluationContext context = new StandardEvaluationContext();

        for (int i = 0; i < parameterNames.length; i++) {
            context.setVariable(parameterNames[i], args[i]);
        }

        Expression expression = parser.parseExpression(distributedLock.key());
        return "LOCK:" + expression.getValue(context, String.class);
    }
}

 

전체적인 내용은 동일하다. 스핀락을 사용해서 분산락을 획득하고, 서비스 특성 상 계좌번호를 기준으로 동시성 제어가 필요하기에 key 로써 계좌번호를 사용하였다. 

    @Override
    @Transactional
    @DistributedLock(key = "#buyOrderRequest.accountNumber")
    public BuyOrderDto createMarketOrder(BuyOrderRequest buyOrderRequest){ ... }

 

실제 어노테이션 사용 시에도 계좌번호를 주입시켜서 분산락에 사용하였다. 

 

하지만 정상적으로 Lock 을 획득하여 실행되지만 실행 결과는 보장이 되지 않는 문제가 있었다. 이는 위에서 찾아본 Synchronized 와 @Transactional 의 관계와 동일한 문제로, AOP 로 실행되는 분산락을 해제하였지만 commit 이 되지 않아 동시성 문제가 발생하는 것이다. 

 

@Slf4j
@Aspect
@Component
@Order(2)
@RequiredArgsConstructor
public class DistributedLockAspect { ... }

 

그래서 단순하게 AOP 실행 시 Order 를 명시해줘서 트랜잭션이 끝난 이후에 Lock 을 해제할 수 있도록 명시해줬다. 

 

그 결과 동시성이 해결되었지만, 성능적으로 우수한 모습을 보이지는 않았다. 

 

// 트랜잭션 커밋 후 락 해제를 위한 콜백 등록
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
    @Override
    public void afterCommit() {
        releaseLock(lockKey, lockValue);
        log.info("Lock released after transaction commit for key: {}", lockKey);
    }

    @Override
    public void afterCompletion(int status) {
        if (status != STATUS_COMMITTED) {
            releaseLock(lockKey, lockValue);
            log.info("Lock released after transaction rollback for key: {}", lockKey);
        }
    }
});

 

별개의 방법으로는 TransactionSynchronizationManager 를 사용해서 트랜잭션 커밋 후 락 해제를 위한 콜백을 등록할 수 있다. 

3) Redisson 분산락

Redisson 라이브러리 같은 경우는 스핀락 방식이 아닌 pub/sub 방식으로 보다 효율적인 분산락 획득 방식을 가지고 있다. 

dependencies {
	...
    implementation 'org.redisson:redisson-spring-boot-starter:3.16.1'
}

@Configuration
public class RedissonConfig {
    @Value("${spring.redis.host}")
    private String redisHost;

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

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

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer()
                .setAddress(REDIS_URL_PREFIX + redisHost + ":" + redisPort);

        return Redisson.create(config);
    }
}

/**
 * 주문 생성 - 시장가 주문
 */
@Override
@Transactional
public BuyOrderDto createMarketOrder(BuyOrderRequest buyOrderRequest){
    final String lockKey = "ORDER:LOCK:" + buyOrderRequest.getAccountNumber();
    final RLock lock = redissonClient.getLock(lockKey);

    try {
        boolean available = lock.tryLock(10, 1, TimeUnit.SECONDS);
        if (!available) {
            System.out.println("redisson lock timeout");
            throw new IllegalArgumentException();
        }

		...
        return BuyOrderDto.fromEntity(saveOrder);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    } finally {
        lock.unlock();
    }
}

 

이렇게 메서드 자체에 RedisLock 을 걸게 되면 동시성에 문제가 발생한다. 하나씩 실행되기는 하지만 역시 트랜잭션이 commit 되기 전에 락이 풀려버리기 때문에 트랜잭션이 확실히 끝난 이후 락을 해제해주는 과정이 필요하다.

 

@Slf4j
@Aspect
@Component
@Order(2)
@RequiredArgsConstructor
public class DistributedLockAspect {
    private final RedissonClient redissonClient;
    private final ExpressionParser parser = new SpelExpressionParser();

    @Around("@annotation(distributedLock)")
    public Object lock(ProceedingJoinPoint joinPoint, DistributedLock distributedLock) throws Throwable {
        String lockKey = obtainLockKey(joinPoint, distributedLock);
        RLock lock = redissonClient.getLock(lockKey);

        try {
            boolean available = lock.tryLock(
                    distributedLock.waitTime(),
                    distributedLock.leaseTime(),
                    TimeUnit.MILLISECONDS
            );

            if (!available) {
                throw new RuntimeException("Failed to acquire lock after " + distributedLock.waitTime() + "ms for key: " + lockKey);
            }

            log.info("Lock acquired for key: {}", lockKey);
            return joinPoint.proceed();

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Lock acquisition was interrupted", e);
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
                log.info("Lock released for key: {}", lockKey);
            }
        }
    }

    private String obtainLockKey(ProceedingJoinPoint joinPoint, DistributedLock distributedLock) {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        String[] parameterNames = signature.getParameterNames();
        Object[] args = joinPoint.getArgs();

        EvaluationContext context = new StandardEvaluationContext();

        for (int i = 0; i < parameterNames.length; i++) {
            context.setVariable(parameterNames[i], args[i]);
        }

        Expression expression = parser.parseExpression(distributedLock.key());
        return "redisson:lock:" + expression.getValue(context, String.class);
    }
}

 

기존 AOP 를 Redisson 으로 대체하였고, Order 어노테이션을 통해 트랜잭션 이후에 락을 해제하도록 순서를 명시해줬다. 

 

그 결과 확실하고 보다 효과적인 동시성 제어가 되는 것을 확인할 수 있었다. 

4. 결론

동시성 제어에서 가장 중요한 점은 트랜잭션이 확실하게 끝난 이후에 락을 해제하는 것이다.

 

그리고 상황에 맞게 DB의 비관적 락, 낙관적 락, Redis 의 분산락을 사용할 수 있겠다.

 

DB 의 비관적 락 같은 경우에는 데이터 일관성이 가장 강력하게 유지되지만 락을 획득하기 위해서 대기하는 경우에도 DB Connection 을 소유하고 있다는 점이 커넥션 풀을 고갈 시킬 수 있다. 

 

Redis 분산락 같은 경우에는 데이터베이스의 부하를 줄일 수 있고 직접적으로 데이터를 잠그는 것이 아니라 race condition 에 진입점을 제어하는 것으로 볼 수 있다. 또한 데이터를 Redis 안에서 관리한다면 데이터베이스 작업 없이 훨씬 빠르게 처리할 수 있으며 이 같은 경우에서는 Lua Script 를 사용해 처리할 수도 있다. 물론 Redis 의 이중화 및 데이터베이스 동기화 작업에도 신경을 써야할 것이다. 

 

현재 개발하고 있는 프로젝트가 금융 도메인이라는 가정 하에 DB 락과 Redis 분산락을 함께 사용할 예정이다. 

5. 요약

1. 동시성 제어에서는 트랜잭션이 끝난 이후 락을 해제해야한다.
2. 여러 가지 방법 중 상황에 맞는 방법을 선택해야 한다. (DB 락, Redis 분산락 ... )