[TIL] 내일배움캠프 78일차 - SQS 동시성 테스트

 

 

오늘의 학습 키워드📚

실행 계획

public class Product {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long productId;

    @Column
    private String productName;

    @Column
    private Integer productStock;

    public void decreaseStock() {
        if (this.productStock > 0) {
            this.productStock -= 1;
        } else {
            throw new IllegalArgumentException("재고가 부족합니다.");
        }
    }
}

Product Entity에서 재고 감소 로직을 추가한다.

    @SqsListener(value = "${cloud.aws.sqs.queue.name}")
    public void sqsListener(@Payload ProductRequest request) {
        Product product = productRepository.findById(request.getProductId()).orElse(null);

        if (product != null) {
            try {
                product.decreaseStock();
                productRepository.save(product);
                log.info("구매 완료. 남은 재고 : " + product.getProductStock());
            } catch (IllegalArgumentException e) {
                log.info(e.getMessage());
            }
        }
    }

SqsListener를 통해 메세지를 순차적으로 받아서 재고 관리를 한다.
요청은 JMeter로 10개의 요청을 동시에 보낸다.

FIFO 큐를 생성해 사용하고 있기 때문에 10개의 요청이 동시에 간다 해도 메세지를 순차적으로 수신해 재고 관리가 자동으로 되는 것을 기대했다.

결과(Fail)

상품의 재고는 3개이고 10개의 요청을 동시에 보내면 3개까지 구매가 되고 그 이후부터는 예외가 발생해 에러 메세지가 발생할 것이라 생각했다.

2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [nio-8080-exec-9] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [nio-8080-exec-4] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [nio-8080-exec-6] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [nio-8080-exec-5] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [nio-8080-exec-3] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [nio-8080-exec-2] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [io-8080-exec-10] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [nio-8080-exec-8] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [nio-8080-exec-7] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T21:46:46.377+09:00  INFO 20964 --- [ReactTest] [nio-8080-exec-1] com.sparta.reacttest.sqs.SqsService      : Sender: 1

2024-04-18T21:46:46.615+09:00  INFO 20964 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 2
2024-04-18T21:46:46.693+09:00  INFO 20964 --- [ReactTest] [ntContainer#0-4] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1
2024-04-18T21:46:46.693+09:00  INFO 20964 --- [ReactTest] [ntContainer#0-5] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1
2024-04-18T21:46:46.694+09:00  INFO 20964 --- [ReactTest] [ntContainer#0-2] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1
2024-04-18T21:46:46.694+09:00  INFO 20964 --- [ReactTest] [ntContainer#0-8] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1
2024-04-18T21:46:46.695+09:00  INFO 20964 --- [ReactTest] [ntContainer#0-6] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1
2024-04-18T21:46:46.697+09:00  INFO 20964 --- [ReactTest] [ntContainer#0-7] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1
2024-04-18T21:46:46.697+09:00  INFO 20964 --- [ReactTest] [ntContainer#0-3] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1
2024-04-18T21:46:46.698+09:00  INFO 20964 --- [ReactTest] [ntContainer#0-9] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1
2024-04-18T21:46:46.698+09:00  INFO 20964 --- [ReactTest] [tContainer#0-10] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1

🤔...?

10개의 요청은 동시에 발생했고 첫 번재 요청은 정상적으로 재고 관리가 되었지만 2번째 요청부터 마지막 요청까지는 같은 재고를 구매하는 문제가 발생했다.

분명 FIFO라고 했는데.....

고민하다 나온 결론은 Queue에서 메세지를 수신하는 속도와 DB에서 작업하는 속도가 다르기 때문에 발생하는 문제라고 생각했다.
Queue에서는 메세지가 들어오면 바로 서버로 보내주는데 서버에서는 DB를 처리하는 시간이 있기 때문에 들어온 메세지들이 결국 서버 내에서는 동시성 제어가 되지 않는 것이라 생각했다.
근데 지금은 해결했지만 아직도 드는 의문은 '대체 왜 첫 번째 요청은 정상적으로 처리되는가?' 이다. 잘 모르겠다 이 부분은...

해결책

1. @Version 어노테이션으로 DB에 낙관적 락 걸기(실패)

SQS로 동시성 제어를 하는게 SQS 사용 목적인데 DB에 락을 거는 것부터 잘못됐지만 우선 에러를 고치고 파악하기 위해 시도해 봤다.
결국 한 번에 들어온 요청이 제대로 처리되지 않는 것이니 DB에 락을 걸어 한번에 들어오더라도 동시성 제어를 또 걸면 되지 않을까? 했었다.

 

반응은 매우 폭발적이었다. 저 기나긴 스크롤 아래가 전부 에러 메세지이다...
재고도 한 번에 2개씩 줄어들고 뭔가 락이 제대로 걸리지 않는 모습을 보였다.

2. Transaction 분리하기(실패)

현재는 SqsService에서 재고 관리까지 전부 담당하고 있다.
이것을 SqsService에서는 메세지 송수신, ProductService에서 트랜잭션을 담당하도록 분리하면 잘 처리될 수 있지 않을까? 했다.

역시나 제대로 제어되지 않았다. 당연하지만 서비스 로직을 분리한다고 해도 결국 메세지를 수신하는 속도와 트랜잭션이 실행되는 속도에는 차이가 있기 때문에 똑같은 현상이 발생한다.

3. 리스너 Config 설정하기(성공!)

계속해서 헤매던 중에 SqsMessageListenerContainerFactory에 configure 프로퍼티로 설정값을 변경할 수 있다는 것을 알았다.

현재 나는 AwsSqsConfig라는 Configuration 클래스에서 리스너 설정을 전부 디폴트로 해놓고 있었다.

    // Listener Factory 설정 (Listener 쪽)
    @Bean
    public SqsMessageListenerContainerFactory<Object> defaultSqsListenerContainerFactory() {
        return SqsMessageListenerContainerFactory.builder()
            .sqsAsyncClient(sqsAsyncClient())
            .build();
    }

근데 SqsMessageListenerContainerFactory 클래스로 들어가면 수많은 프로퍼티들을 설정할 수 있는데 그중 configure라는 프로퍼티를 통해 SQS 메세징 처리에 있어 설정 값들을 커스텀할 수 있다.

        public Builder<T> configure(Consumer<SqsContainerOptionsBuilder> options) {
            this.optionsConsumer = options;
            return this;
        }

public interface ContainerOptionsBuilder<B extends ContainerOptionsBuilder<B, O>, O extends ContainerOptions<O, B>> {
    B maxConcurrentMessages(int maxConcurrentMessages);

    B maxMessagesPerPoll(int maxMessagesPerPoll);

    B maxDelayBetweenPolls(Duration maxDelayBetweenPolls);

    B pollTimeout(Duration pollTimeout);

    B listenerMode(ListenerMode listenerMode);

    B componentsTaskExecutor(TaskExecutor taskExecutor);

    B acknowledgementResultTaskExecutor(TaskExecutor taskExecutor);

    B listenerShutdownTimeout(Duration shutdownTimeout);

    B acknowledgementShutdownTimeout(Duration acknowledgementShutdownTimeout);

    B backPressureMode(BackPressureMode backPressureMode);

    B acknowledgementInterval(Duration acknowledgementInterval);

    B acknowledgementThreshold(int acknowledgementThreshold);

    B acknowledgementMode(AcknowledgementMode acknowledgementMode);

    B acknowledgementOrdering(AcknowledgementOrdering acknowledgementOrdering);

    B messageConverter(MessagingMessageConverter<?> messageConverter);

    O build();

    B createCopy();

    void fromBuilder(B builder);
}

설정값이 정말 많은데... 대표적으로 사용하는 몇 가지가 있다.

• maxConcurrentMessages : 동시 처리할 메세지의 최대 개수
• maxMessagesPerPoll : 메세지를 폴링하는 최대 개수
• acknowledgementMode : 메세지 처리 모드(항상 삭제, 성공 시에만 삭제, 전부 수동)

이 외에 설정값들도 물론 많이 사용되겠지만 지금 당장 볼만한 건 위의 3개인 것 같다.
특히 maxMessagesPerPoll 옵션이 굉장히 중요한데 메세지를 수신할 때 최대 몇 개의 메세지를 가져올 건지를 설정하는 항목이다.

    // Listener Factory 설정 (Listener 쪽)
    @Bean
    public SqsMessageListenerContainerFactory<Object> defaultSqsListenerContainerFactory() {
        return SqsMessageListenerContainerFactory.builder()
            .sqsAsyncClient(sqsAsyncClient())
            .configure(
                option -> option
                    .maxMessagesPerPoll(1)
            )
            .build();
    }

나는 재고 동시성 제어를 위해 하나씩 처리하는 것이 목적이므로 값을 1로 주고 실행해 봤다.
이것 외에도 acknowledgementMode를 설정해 트랜잭션을 처리하는 동안 메세지를 점유하는 방법도 고려해 볼 만하다.

2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [io-8080-exec-10] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [nio-8080-exec-2] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [nio-8080-exec-7] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [nio-8080-exec-9] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [nio-8080-exec-1] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [nio-8080-exec-8] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [nio-8080-exec-4] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [nio-8080-exec-5] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [nio-8080-exec-6] com.sparta.reacttest.sqs.SqsService      : Sender: 1
2024-04-18T22:27:17.835+09:00  INFO 16580 --- [ReactTest] [nio-8080-exec-3] com.sparta.reacttest.sqs.SqsService      : Sender: 1

2024-04-18T22:27:18.071+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 2
2024-04-18T22:27:18.132+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 1
2024-04-18T22:27:18.147+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 구매 완료. 남은 재고 : 0
2024-04-18T22:27:18.151+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 재고가 부족합니다.
2024-04-18T22:27:18.155+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 재고가 부족합니다.
2024-04-18T22:27:18.159+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 재고가 부족합니다.
2024-04-18T22:27:18.163+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 재고가 부족합니다.
2024-04-18T22:27:18.167+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 재고가 부족합니다.
2024-04-18T22:27:18.171+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 재고가 부족합니다.
2024-04-18T22:27:18.174+09:00  INFO 16580 --- [ReactTest] [ntContainer#0-1] com.sparta.reacttest.sqs.SqsService      : 재고가 부족합니다.

드디어 성공했다!!!
10개의 구매 요청이 동시에 발생했고 먼저 처리된 3명은 재고를 확보했지만 이후 7명은 동시에 구매 요청을 했어도 동시성 제어에 걸려 나중에 처리가 되었기 때문에 재고 확보를 할 수 없었다.

이렇게 하면 만약 특가 상품을 기획해 순간 트래픽이 몰려 수많은 사람들이 동시에 구매 요청을 한다 해도 SQS를 통해 정해진 재고만 판매할 수 있을 것이다.

---

오늘의 회고💬

SQS를 이용해 주문 및 결제 처리는 완료했는데 결제 성공 or 실패 부분이 SQS 메세지 수신 콜백 함수여서 서버에서 클라이언트에 직접 전달해 줄 수 없었다. 그래서 SSE를 사용하기로 했고 현재 구현 중에 있다.

 

내일의 계획📜

SSE로 결제 성공, 실패에 대한 이벤트를 수신하고 프론트에 대한 부분을 좀 더 다듬어야 할 것 같다.