[AWS] SQS 가시성제한 0초 설정

개요

Server가 여러 대의 Instance로 구성되어 있을 때 동기화를 위해 모두 SQS 메시지를 Consume 하게 하고 싶습니다.

예를 들어 @SqsListener를 가지는 Instance가 10대라면 동일한 메시지에 대해서 10번의 Consume이 발생하는 것입니다.

이를 테스트하기 전 다음과 같은 가설을 세워보았습니다.

 

가설 1 - 가시성 제한 0초

위의 요구사항을 구현하기 위해 가장 중요한 것은 가시성제한입니다.

가시성 제한은 기본적으로 30초이며 하나의 Consumer가 메시지를 읽고 있으면 30초간 다른 Consumer는 해당 메시지를 읽을 수 없습니다.

이 가시성 제한을 0초로 설정하면 모든 Consumer가 메시지를 읽을 수 있지 않을까요?

 

가설 2 - 네트워크 지연

하지만 네트워크 지연은 불확실하니 모든 Consumer가 동일한 시간대에 메시지를 읽을 순 없을 것 같습니다.

SQS의 경우 메시지를 읽고 나서 삭제하는 정책을 가진다면 다음과 같은 일이 일어날 수 있을 것 같습니다.

• 메시지를 읽고 처리하는 시간이 0.1초 걸린다고 가정
• A Consumer에서 메시지를 0.1초 만에 읽고 삭제
• B Consumer는 네트워크 지연으로 이미 삭제된 메시지를 읽어오지 못하지 않을까?

 

가설 3 - Consumer들의 동시 삭제?

• A Consumer에서 메시지를 읽고 삭제를 했다
• B Consumer에서는 메시지를 읽고 처리하고 삭제를 하려고 하는데 A Consumer에서 이미 삭제가 되면 어떻게 동작할까?

 

가설 4- 메시지를 삭제하지 않으면 어떻게 될까?

SQS를 사용하게 되면 메시지 삭제 정책을 설정할 수 있습니다.

예를 들어 항상, 성공했을 때, ack를 승인했을 때 삭제할 수 있습니다.

 

이때 ack를 승인하지 않아 메시지를 삭제를 하지 않으면 어떻게 될까요?

SQS는 최대 메시지 보존 기간(최대 14일)보다 오래 대기열에 있는 메시지를 자동으로 삭제합니다.

 

하지만 지속적으로 메시지를 요청하니.. 서버에 부하가 발생할 것 같습니다.

 

 

테스트 시작

1대의 Instance를 띄우고 하나의 SQS에 테스트 메시지 전송

 

무수한 Consume..

짧은 시간에 40개 정도를 하나의 Container에서 Consume 하는 모습을 로깅으로 확인할 수 있었습니다.

내가 원한 건 Contianer1대에 1건인데..

 

SQS 설정 확인

@Bean
  fun defaultSqsListenerContainerFactory(
    sqsAsyncClient: SqsAsyncClient,
    queueContainerTaskExecutor: AsyncTaskExecutor,
  ): SqsMessageListenerContainerFactory<Any> {
    return SqsMessageListenerContainerFactory
      .builder<Any>()
      .configure { options ->
        options
          .acknowledgementMode(AcknowledgementMode.ON_SUCCESS)
          .maxMessagesPerPoll(10)
          .pollTimeout(Duration.ofSeconds(5))
          .componentsTaskExecutor(queueContainerTaskExecutor)
          .queueNotFoundStrategy(QueueNotFoundStrategy.FAIL)
      }
      .sqsAsyncClient(sqsAsyncClient)
      .build()
  }

AcknowledgementMode는 메시지 처리에 성공했을 때 삭제하도록 했습니다.

가시성 제한이 0초여서 메시지 처리 전에 계속 가져오는 건가 싶어서 가시성 제한을 높여보겠습니다.

 

SQS 가시성 제한 0초 -> 10초

가시성 제한을 10초로 설정해 보고 다시 메시지를 발행 모니터링을 수행합니다.

1건만 consume 하는 모습을 볼 수 있습니다.

하지만 여러 Instance에서 Consume을 하기 위해서는 가시성 제한이 0초로 되어야 할 것 같습니다.

 

메시지 수신 대기 시간을 0초 -> 5초로 변경

수신대기 시간을 0초로 설정하면 짧은 폴링이 발생합니다.

사용할 수 없는 메시지가 없는 경우 계속 요청하는 것이 아니라 5초의 기간을 가지고 롱 폴링을 시도하도록 변경해 보겠습니다.

여전히 많은 메시지가 Consume 됩니다.

 

 

 

의심되는 부분 하나 - SQSAsyncClient

분산된 SQS Server

SQSAsyncClient를 현재 사용하고 있으며, SQS의 아키텍처는 분산되어 중복된 메시지가 저장되고 있습니다.

이로 인해 여러 개의 분산된 서버에서 메시지를 가져오기 때문에 이런 현상이 발생하는 것 같다고 생각됩니다.

즉, 메시지가 처리되어 삭제되기 전에 이미 여러 분산된 서버에서 Consume을 해버리기 때문에 발생하는 것 같습니다.

 

해결방법으로 시도해 보려고 생각나는 것들입니다.

1. SQSAsync -> SQSClient로 변경해 보자 -> SQSAsync만 SqsMessageListenerContainerFactor에 적용가능 -> 변경불가능

2. 메시지를 Consume 하자마자 MANUAL로 ACK를 날려 지우자 -> 효과 없음

3. 가시성 제한을 컨테이너에 걸어줘 보자 그러면 인스턴스마다 가시성제한을 거는 효과를 얻을 수 있지 않을까? 

//java doc 설명: 컨테이너에서 검색한 메시지에 대한 메시지 표시 여부를 설정합니다.
.messageVisibility(Duration.ofSeconds(30L))

드디어 Instance가 1건일 때 단건이 나옵니다.

 

Instance 2대로 테스트 시작

인스턴스를 2대로 구성하고 테스트를 수행하여도 여전히 한대의 메시지만 수신됩니다.

이로써 SQS를 통해서는 여러 인스턴스에 동기화가 안된다는 것을 깨닫게 되었습니다.

혹시나 하는 방법을 아시는 분이 계신다면 댓글로.. 부탁드립니다

 

결론

최종적으로 인스턴스별 동기화를 하고 싶다면 redis의 pub/usb이나 kafka를 활용하여 Instance의 POD ID 등을 통해 Consumer 그룹을 독립적으로 구성하고 메시지를 Consume 하여 동기화를 수행할 수 있을 것 같습니다.