이벤트 발행/구독 모델을 활용한 실시간 데이터 동기화 feat. AWS SNS, SQS

🎯 프로젝트 목표

앞서 기술회고에서 우리 서비스가 다른 서비스와 실시간으로 데이터가 동기화되고 있다는 언급을 하였다.

가끔 예측하지 못한 이유로 이러한 동기화가 제대로 작동하지 않는 경우가 있다. 빠른 후속조치를 위해 상태 변화 전송이 누락된 사용자에 대해 이벤트를 발행하고 데이터를 완전히 동기화시키는 기능이 필요해졌다. 

사용자 아이디의 리스트를 받아 해당 사용자의 상태 변화를 트래킹 할 수 있는 이벤트를 전송하는 것을 목표로 한다. 

 

🤔 구현 방식

이벤트 발행을 트리거할 수 있는 API를 디자인한다.

사용자의 정보를 담은 적절한 이벤트 객체를 생성하고 발행한다. 

이벤트가 타 서비스에서 잘 소비되고 있는 지 확인한다. 

⚒️ 기술 스택

이 글은 이벤트 발행과 구독, 소비하는 패턴의 흐름을 중점적으로 다루는 글이다. 

우리 프로젝트의 이벤트 발행 - 구독 시스템은 AWS SNS, SQS를 사용한다. 

SNS (Simple Notification Service) 

• 메시지를 여러 구독자에게 발행하기 위해 사용하는 AWS 서비스이다. 
• 이 메시지를 구독하는 구독자는 SQS 큐뿐만 아니라 이메일, HTTP 등 다양한 형태가 있다. 
• SNS Topic은 구독자가 구독하는 하나의 단위 (채널)이다. 슬랙이 SNS처럼 하나의 시스템이라면, 각 슬랙 채널은 SNS Topic에 해당한다. 해당 채널에 관심이 있는 사람 (구독자)만 해당 채널에 들어가는 것처럼, 구독도 SNS Topic 단위로 이루어진다.

SQS (Simple Queue Service)

• 메시지를 큐에 저장하고 비동기로 메시지를 Consumer가 꺼내가기 위해 사용하는 AWS 서비스이다. 
• 특정 SNS Topic을 구독하고, 메시지가 발행되어 큐에 저장되었을 때 별도의 서비스에서 메시지를 처리한다.

 

왜 SNS, SQS를 사용하는가? 

기술적으로는 사실 SNS나 SQS가 없어도 이벤트 발행과 구독이 가능하다. HTTP 형태로 다른 서비스에 직접 데이터를 Push할 수 있고, 혹은 제3의 Gateway에 데이터를 보내놓은 다음 다른 서비스가 원할 때 Pull 방식으로 가져갈 수도 있다.

하지만 왜 SNS, SQS 를 사용해야할까? 바로 가용성 (Scalable) 때문이다. 

이렇게 여러 서비스가 우리 서비스에 의존하게 된다면, 의존하는 서비스가 늘 수록 우리 서비스의 변경 지점이 많아진다. 

이 책임을 SNS가 지게되면 어떨까? SNS가 구독자를 직접 정의해야하긴 하지만, 우리 서비스와의 의존성은 분리되므로 괜찮아보인다.

하지만 SNS Topic이 여러 개가 되면 어떨까? 

이렇게 SNS Topic마다 중복된 서비스를 직접 등록해야하는 일이 생길 것이다. 이벤트 구독의 유무가 변할 때마다 중복된 정보를 수정해주어야하는 비용이 생긴다. 

 

이렇게 SNS와 SQS의 조합을 사용하면 상대적으로 자주 변하는 이벤트 발행・구독 정보에 최소화된 변경 지점으로 아키텍처를 수정할 수 있게 된다.  

---

👩‍💻 구현 과정

이벤트를 발행하는 코드는 간단하다.

AmazonSNS 를 사용해 코드 상에서 이벤트 발행을 트리거한다. 

• topicArn: 전송할 SNS 토픽의 이름이다.
• message: 전송할 이벤트의 내용을 문자열 형태로 담는다.

원하는 토픽과, 메시지를 담아서 해당 함수를 호출해주면 된다.

 

AWS SNS Topic 페이지를 들어가면 ARN을 확인할 수 있다.

아래 Subscriptions 탭에서는 SNS에 구독 중인 SQS의 목록을 확인할 수 있다.

 

Subscription 중 하나를 클릭하면 SQS의 상세 정보를 볼 수 있다. 

  

 

발행된 메시지를 구독하는 쪽에서는 어떻게 구현할 수 있을까?

우리가 보낸 메시지를 구독하는 서비스의 프로젝트로 찾아가 관련 코드를 찾아보았다.

 

먼저, 위의 구독 정보에서 명시된 큐 이름이 설정파일에 등록되어있다. 

# application.yml 
eventbus: 
	env_name_of_queue: queue_name_in_sqs

 

@JmsListener를 사용하면 JMS 메시지를 리스닝하는 메시지를 정의할 수 있다. 

 

JMS 메시지란?

Java Message Service라는 뜻, 자바 어플리케이션에서 메시지를 생성하거나 읽을 수 있는 기능을 지원한다.

@JmsListener(destination = "\${eventbus.env_name_of_queue}")
fun listen(message: SQSTextMessage): Nothing? = listen<MessageHolder>(message) { 
   	// do something
}

 

우리가 JSON 형태로 보낸 메시지를 MessageHolder라는 객체로 전달받을 수 있다. 

@JmsListener의 destination에는 이 함수가 수신 대기해야하는 대상의 이름이 들어간다. 

 

이렇게 스프링에서 제공하는 기능을 사용하면 이벤트가 발행되어 구독하는 큐에 이벤트가 수신되었을 때,

JmsListenerContainer가 내부 리스너를 생성해 큐를 구독한다. @JmsListener는 내부적으로 JMS 메시지를 지속적으로 감시하는 리스너 컨테이너를 생성한다.

 

리스너 컨테이너가 필요한 이유? 

백그라운드에서 리스너가 계속 돌아가고 있어야 등록된 desitnation, 즉 큐에 이벤트가 생성되었을 때 소비할 수 있다. 

 

이렇게 이벤트 발행하는 서비스에서는 AmazonSns 라이브러리를 사용해서, 

SNS Topic을 구독하고 있는 서비스에서는 JMS 라이브러리를 사용해 이벤트를 백그라운드에서 수신한다.

 

---

✏️ 느낀 점

우리 팀 코드 베이스에서 어떻게 이벤트를 전달하고, 외부 서비스에서 어떻게 우리 이벤트를 소비하는 지 이해할 수 있게 된 프로젝트였다.

이를 통해 이벤트 발행/구독 모델과 AWS 서비스 사용에 익숙해질 수 있었다. 해당 태스크를 진행하면서 시니어 개발자분이 이벤트 발행/구독 구조에 대해 설명해주시고, 왜 이런 모델이 필요한지 설명해주셨다. 당연하게 받아들이고 있던 것에 '왜'라는 질문을 던지는 자세가 중요함을 한번 더 인지할 수 있었다. 단순히 이벤트를 전달하는 API를 추가하는 것에서 끝나는 게 아니라, 다른 서비스에서 어떻게 이벤트를 전달받고 있는 지 알아보는 기회가 되었다. 

 

✏️ 참고

https://spring.io/guides/gs/messaging-jms

Getting Started | Messaging with JMS