SQS + Lambda를 이용하여 사용자 수에 무관한 푸시알림 전송 환경 구축하기

버전
Spring Boot 3
AWS SDK for java v2
firebase-admin 6.5.0

0. 도입 배경

기존 구조

기존에는 푸시 알림 요청에 따라서 서버 내에서 메세지를 생성하고 Firebase로 메세지를 전송했다.

 

하지만, 여기서 푸시 알림을 보낼 사용자가 만명, 10만명, 100만명이 된다면...

 

모든 사용자에 대한 알림을 일괄적으로 처리해야 한다.

 

일괄 처리를 위한 Firebase의 라이브러리를 보자.

 

com.google.firebase.FirebaseMessaging.sendAll()

 

먼저 한꺼번에 보내는 sendAll은 Deprecate되었고, 대신 sendEach를 써야한다.

 

com.google.firebase.FirebaseMessaging.sendEach()

 

sendEach()는 각 토큰별 여러 메세지를 보낼 수 있지만, 각 메세지마다 별도의 HTTP 요청을 만들어낸다.

 

10만개의 요청이면 10만개의 HTTP 요청이 생성된다는 것이다.

 

심지어, message의 양은 500개가 제한이기 때문에 500개씩 나눠서 호출해야 한다.

 

 

 

sendEachAsync()는 비동기로 동작하지만 리소스를 사용하는 것은 마찬가지이다.

 

 

 

개선한 구조

 

따라서 나는 토큰과 푸시 알림 내용을 담은 메세지를 SQS에 적재하고,

 

SQS에 메세지가 도착하면 Lambda가 트리거되어 파이썬 함수가 실행되는 구조로 개선해 보려고 한다.

 

firebase_admin 모듈이 설치되어 있는 AWS Lambda에서 푸시 알림을 처리한다.

 

기존 구조와는 다르게, 전송하려는 token의 양이 많더라도 제한이 없고, token양에 상관 없이 서버 입장에서는 무조건 1번의 요청으로만 해결하려고 한다.

 

만 명의 사용자에게 푸시알림을 전송해도 한 번의 HTTP 요청으로 해결한다는 것이다.

 

해당 구조는 푸시 알림 전송이 완전히 분리되어, 다음과 같은 효과를 기대해볼 수 있다.

 

• 확장성 (Scalability)
  • AWS SQS는 높은 처리량과 확장성을 제공한다. 메시지 양이 증가해도 쉽게 처리할 수 있다.
  • Lambda 함수는 자동으로 스케일링되어 동시에 많은 푸시 알림을 처리할 수 있다.
• 분산 처리 (Decoupling)
  • 메인 서버와 푸시 알림 처리 로직을 분리함으로써 시스템의 결합도를 낮추고 유지보수성을 높인다.
  • 서버의 부하를 줄이고, 푸시 알림 실패가 서버의 다른 기능에 영향을 주지 않는다.
• 안정성과 내구성 (Reliability and Durability)
  • SQS는 메시지를 안전하게 저장하고 관리한다. 일시적인 네트워크 문제나 서비스 중단 시에도 메시지 손실 없이 재시도할 수 있다.
• 모니터링과 로깅 (Monitoring and Logging)
  • AWS 서비스를 사용함으로써 CloudWatch 등을 통해 쉽게 모니터링하고 로그를 관리할 수 있다.
• 비동기 처리 (Asynchronous Processing)
  • 푸시 알림 전송을 비동기적으로 처리함으로써 메인 애플리케이션의 응답 시간을 개선할 수 있다.
• 재시도 매커니즘 (Retry Mechanism)
  • SQS의 가시성 타임아웃과 DLQ를 이용해 실패한 메시지를 효과적으로 재처리할 수 있다.

 

 

 

 

1. IAM 생성

먼저 SQS에 메세지를 보낼 권한이 필요한데, 이를 IAM에서 만들어준다.

 

새로운 사용자를 생성하면서, AmazonSQSFullAccess 권한을 가지고 있는 사용자를 만들어준다.

 

사용자가 생성됐다면, 엑세스 키를 만들어준다.

 

 

엑세스 키는 테스트 시에만 "AWS 외부에서 실행되는 애플리케이션" 으로 만들어주고,

 

실제 배포한 환경이 AWS EC2 내라면 "AWS 컴퓨팅 서비스에서 실행되는 애플리케이션"의 키를 사용해주는게 좋다.

 

 

엑세스 키를 생성하면 1회성으로 엑세스 키와 비밀 엑세스 키를 보여주니 고이 모셔둔다.

 

 

 

 

2. SQS (Simple Queue Service) 인스턴스 생성

 

이제 SQS 에서 대기열을 새로 생성해주면 된다.

 

 

큐에 들어갈 메세지의 성격에 따라 구성을 해주면 되는데,

 

내 서비스에서 사용할 메세지는 1시간이 지나면 의미 없는 메세지이기 때문에 보존 기간을 1시간으로 설정했다.

 

 

여기서 대기열에 메세지를 보낼 수 있는 사용자에 방금 만들었던 IAM의 Role을 넣어주면 된다.

 

 

ARN Role은 IAM 사용자에 들어가서 복사할 수 있다.

 

3. Lambda 함수 생성

 

 

다음으로 Lambda 함수를 생성하면 되는데, 런타임 언어는 파이썬으로 하는게 가장 좋다.

 

파이썬의 firebase_admin 모듈이 가장 가볍고 설정하기 간편하기 때문이다.

 

 

 

 

firebase-admin 모듈 설치

 

다음으로 firebase admin 모듈을 함수에 설치해보자.

 

 

먼저 python으로 된 빈 디렉토리를 생성한다.

 

‼️ Lambda함수에서 모듈로 인식하기 위해서는 최상위 폴더의 이름이 꼭 python이어야 한다!

 

 

 

pip을 이용하여 firebase-admin 소스를 가져온다.

$ pip install firebase-admin -t {경로}/python

 

 

 

소스가 설치되었다면, 그대로 압축해준다.

 

 

AWS Lambda로 돌아와  계층을 새로 생성해준다.

 

 

 

 

계층에는 .zip 파일을 업로드하여 생성할 수 있다.

 

방금 압축했던 zip파일을 업로드하고, Lambda함수의 호환 아키텍처와 언어를 선택해주면 된다.

 

 

성공적으로 생성되었다면 버전을 확인한다. (처음 만들었다면 1일 것이다.)

 

 

이제 함수로 돌아와, 새로운 계층을 생성해준다.

 

 

방금 만들었던 계층을 추가해준다.

 

이로써 해당 함수에 firebase_admin 모듈이 설치되었다.

 

 

 

 

트리거 추가

 

이제 해당 함수가 실행되기 위한 트리거로, SQS할당해주면 된다.

그 전에 권한을 연결해야 하는데,

 

해당 함수의 구성-권한에 들어가서 Role 로 들어간다.

 

 

정책 연결을 선택한다.

 

AWSLambdaSQSQueueExecutionRole를 추가해준다.

 

정책이 추가되었다면 함수로 돌아와 트리거를 추가한다.

 

 

전에 만들었던 SQS를 연결해주면 된다.

 

트리거가 연결되었다.

 

이제 해당 SQS 큐에 메세지가 들어오면 Lambda 함수로 메세지가 전달될 것이다.

함수 코드 작성

이제 해당 함수로 메세지가 전달되면 실행될 코드를 작성하면 된다.

 

 

먼저 해당 폴더에 firebase 프로젝트에서 받은 키 정보 (json)파일을 넣어두어야 한다.

 

import firebase_admin as admin
from firebase_admin import credentials
from firebase_admin import messaging

# Firebase 계정 정보 로드
credential = credentials.Certificate('waitherAccountServiceKey.json')
admin.initialize_app(credential)

# 메인 함수
def lambda_handler(event, context):
    parsed_payload = parse_sqs_message(event)
    return send_notification(parsed_payload)

# 푸시 알림 전송
def send_notification(payload):
    responses = []
    for token in payload['tokens']:
        message = messaging.Message(
            notification=messaging.Notification(
                title=payload['title'],
                body=payload['content']
            ),
            token=token
        )
        response = messaging.send(message)
        responses.append(response)
    return responses

# SQS 메시지 파싱
def parse_sqs_message(sqs_message):
    title = sqs_message['Records'][0]['messageAttributes']['title']['stringValue']
    tokens = sqs_message['Records'][0]['messageAttributes']['tokens']['stringValue']
    tokens = tokens[1:-1].split(", ")
    content = sqs_message['Records'][0]['body']
    return {"title": title, "content": content, "tokens": tokens}

 

메세지가 전송되면 lambda_handler 함수가 실행되는데, 받은 메세지(event)를 파싱해서 firebase_admin 모듈로 메세지를 전송한다.

 

안에 내용 중 tokens를 슬라이싱해서 split 하는 과정이 있는데, 그 이유는 뒤에서 설명하겠다.

 

SQS로 받는 이벤트는 보통 다음과 같은 형식이다.

 

{
  "Records": [
    {
      "messageId": "{메세지 아이디}",
      "receiptHandle": "{SQS 메세지 고유 식별자}",
      "body": "{메세지 바디}",
      "attributes": {
        "ApproximateReceiveCount": "10",
        "SentTimestamp": "1724855187392",
        "SenderId": "AIDAYS2NS75TM4GSXUFYB",
        "ApproximateFirstReceiveTimestamp": "1724855189392"
      },
      "messageAttributes": {
        "title": {
          "stringValue": "테스트 제목입니다.",
          "binaryValue": null,
          "stringListValues": [],
          "binaryListValues": [],
          "dataType": "String"
        },
        "token": {
          "stringValue": "[token1, token2, token3]",
          "binaryValue": null,
          "stringListValues": [],
          "binaryListValues": [],
          "dataType": "String"
        }
      },
      "md5OfMessageAttributes": "{속성 무결성 검증용 MD5 해시값}",
      "md5OfBody": "{바디 무결성 검증용 MD5 해시값}",
      "eventSource": "aws:sqs",
      "eventSourceARN": "{Role ARN}",
      "awsRegion": "ap-northeast-2"
    }
  ]
}

 

큰 틀로 나누면 메세지 속성(messageAttributes)과 body로 나뉘는데,

 

나는 메세지 속성에 사용자 토큰 값들과 푸시알림 제목을 넣을 것이고, body에는 푸시알림 내용을 넣을 것이다.

 

해당 내용을 파싱하는게 parse_sqs_message 함수의 내용이다.

 

 

 

4. 서비스 로직 구현

이제 푸시알림을 전송하는 서버쪽을 구현해보자.

 

Depdency(gradle)

implementation platform('software.amazon.awssdk:bom:2.21.20')
implementation 'software.amazon.awssdk:sqs'

 

라이브러리는 aws-sdk for java 버전 2를 사용할 것이다.

 

application.yml

cloud:
  aws:
    credentials:
      access-key: [access key]
      secret-key: [secret access key]
    stack:
      auto: false
    sqs:
      queue:
        message-delay-seconds: [message-delay-seconds]
        url: [SQS Queue URL]

 

 

AWSConfig

@Configuration
public class AwsConfig {
    @Value("${cloud.aws.credentials.access-key}")
    private String accessKey;
    @Value("${cloud.aws.credentials.secret-key}")
    private String secretKey;

    //Credential
    private StaticCredentialsProvider createAwsCredentialsProvider() {
        AwsBasicCredentials basicAWSCredentials = AwsBasicCredentials.create(this.accessKey, this.secretKey);
        return StaticCredentialsProvider.create(basicAWSCredentials);
    }

    //SQS Async Client
    @Bean
    public SqsAsyncClient sqsAsyncClient() {
        return SqsAsyncClient.builder()
                .region(Region.AP_NORTHEAST_2) //서울 Region
                .credentialsProvider(createAwsCredentialsProvider())
                .build();
    }
}

 

AWSConfig에는 SqsAsyncClient를 빈으로 등록해준다. 

 

SqsClient는 동기 방식, SqsAsyncClient는 비동기 방식의 클라이언트이다.

 

임시 자격 증명 객체(StaticCredentialProvider)는 처음 IAM에서 생성했던 엑세스 키의 access key, secret access key로부터 생성한다.

 

 

AWSSqsProperties

@Getter
@Component
public class AwsSqsProperties {
    @Value("${cloud.aws.sqs.queue.url}")
    private String queueUrl;
    @Value("${cloud.aws.sqs.queue.message-delay-seconds}")
    private Integer messageDelaySecs;
}

 

SQS에 연결할 설정들을 미리 컴포넌트로 등록해 두었다.

 

 

AwsSqsUtils

@Slf4j
@RequiredArgsConstructor
@Component
public class AwsSqsUtils {

    private final SqsAsyncClient sqsAsyncClient;
    private final AwsSqsProperties awsSqsProperties;

    public void sendMessage(SqsMessageDto messageDto) {
        Map<String, MessageAttributeValue> attributes = createAttributes(messageDto.tokens(), messageDto.title());

        SendMessageRequest request = createRequest(messageDto, attributes);

        try {
            CompletableFuture<SendMessageResponse> future = sqsAsyncClient.sendMessage(request);
            future.whenComplete((sendMessageResponse, throwable) -> {
                if (throwable == null) {
                    log.info("[SQS Async Client] 메세지 전송 성공 Status ---> {}", sendMessageResponse.sdkHttpResponse().statusCode());
                    log.info("[SQS Async Client] 메세지 전송 성공 ID ---> {}", sendMessageResponse.messageId());
                } else {
                    log.error("[SQS Async Client] 메세지 전송 실패 ---> {}", throwable.getMessage());
                }
            });

        } catch (SqsException sqsException) {
            log.error("[SQS] SQS 메세지 전송 실패 --> {}", sqsException.getMessage());
        }
    }

    private SendMessageRequest createRequest(SqsMessageDto messageDto, Map<String, MessageAttributeValue> attributes) {
        return SendMessageRequest.builder()
                .queueUrl(awsSqsProperties.getQueueUrl())
                .delaySeconds(awsSqsProperties.getMessageDelaySecs())
                .messageAttributes(attributes)
                .messageBody(messageDto.content())
                .build();
    }

    public void sendMessages(List<String> tokens, String title, String content) {
        sendMessage(new SqsMessageDto(tokens, title, content));
    }

    private static Map<String, MessageAttributeValue> createAttributes(List<String> tokens, String title) {
        Map<String, MessageAttributeValue> attributes = new HashMap<>();
        attributes.put("tokens", MessageAttributeValue.builder().stringValue(tokens.toString()).dataType("String").build());
        attributes.put("title", MessageAttributeValue.builder().stringValue(title).dataType("String").build());
        return attributes;
    }

}

 

이제 메세지를 전송하는 함수를 구현하면 된다.

 

비동기 방식의 AsyncSqsClient를 사용했기 때문에, 응답을 CompletableFuture로 받아야한다.

 

catch하는 SqsException은 AWS에서 지원하는 전송 실패의 예외들이 담겨있고, throwable에 들어오는 예외들은 전송 후 실패 예외들을 담고있다.

 

여기서 여러 token이 담겨있는 리스트를 넣기 위해서 Attribute builder에 리스트를 추가하는 stringListValues 함수가 있지만

 

StringListValues는 아직 구현되지 않았기 때문에 사용하지 못한다 🫠

 

따라서 리스트를 toString으로 넣고 파이썬으로 파싱하는 과정으로 할 수 밖에 없었다.

 

이제 해당 Util을 사용해서 메세지를 전송하면 되겠다.

 

 

5. 추가

 

AWS CloudWatch에서 Lambda 함수의 실행 로그를 확인해볼 수 있다.

 

 

Lambda에서 이벤트를 직접 생성하여 테스트해볼 수 있다.

 

 

추가로, 더욱 안정성 있는 서비를 위해 SQS의 DLQ(Dead Letter Queue)를 운용할 수도 있겠다.

 

 

 

6. 성능 개선 테스트

 

위 구조의 성능은 눈에 띄게 개선된다.

 

토큰이 몇 개라도 요청이 한 번 실행되기 때문이다.

 

리소스 사용 차이를 보기 위해 동기 방식으로 바꾸고 테스트를 진행했다.

SQS + Lambda 기준, 토큰 1000개 전송

 

기존 스프링 Firebase-admin 라이브러리 사용, 토큰 1000개 전송

 

 

 

사용자 1000명에게 푸시알림을 전송하는 테스트를 진행했을 때, 약 98%(2620ms -> 36ms)가 향상되었다.

 

1000명을 테스트 해보았지만, 사용자가 많아지면 많아질수록 성능 차이는 확실하게 더 벌어질 것이다.

 

이로써 사용자 수에 무관한 푸시알림 전송 환경을 구축해 보았다.

 

 

 

 

 

참고
https://cabi.oopy.io/1d1010b2-b287-43af-a2c1-bab35cdf1dd8