[AWS] SQS로 비동기로 요청 처리하기

내가 현재 구현중인 프로젝트는 '수강신청'을 받는 프로그램이다. 

 

따라서 수강신청이 몰리는 시간대(수강신청 오픈 직후로부터 1시간동안)에는 비동기로 요청을 처리할 수 있도록 할 것이다.

 

1.  AWS 설정

1.1 SQS 권한 IAM 사용자 생성

IAM > 사용자 > 사용자 생성

"AmazonSQSFullAccess" 정책을 가진 IAM 사용자를 생성해준다.

 

1.2 SQS 대기열 생성

Amamzon SQS > 대기열 > 대기열 생성

표준 큐를 선택해서 만들었다.

 

초기에는  표준큐로 구현했지만, 수강신청 특성상 순서가 중요하다고 생각하여 FIFO 큐로 변경하였다.

 

 

암호화는 "비활성화"를 선택하고, 

액세스 정책은 "지정된 AWs 계정, IAM 사용자 및 역할만"을 선택했다.

 

안에 값에는, 위에서 생성한 sqs권한 사용자의 ARN 값을 붙여넣기 하면 된다.

 

나머지는 기본 설정으로 두었다.

 

1.3.1 EC2에 IAM 설정 (EC2에서 서버 실행 시)

만약 요청을 보내고자하는 백엔드 서버가 EC2에서 돌아가고 있다면,

EC2에 IAM 권한을 주어야 정상적으로 요청을 보낼 수 있다.

 

EC2 > 인스턴스 > 작업 > 보안 > IAM 역할 수정 

 

1.3.2 IAM 사용자 프로필 설정(로컬에서 서버 실행 시)

만약 요청을 보내고자하는 백엔드 서버가 로컬에서 돌아가고 있다면,

IAM 사용자 프로필을 설정해야 정상적으로 요청을 보낼 수 있다.

 

먼저 로컬 환경에 aws-cli를 설치해야한다.

brew install awscli

 

IAM > 사용자 > sqs사용자 > 액세스 키 만들기

aws cli에서 사용할 것이기 때문에, CLI 액세스 키를 선택한다.

 

이렇게 키를 생성하면, Access Key ID와 Secret Access Key가 나오는데, 해당 페이지를 벗어나지 말고 

로컬 환경에서 아래 명령어를 입력한다.

aws configure --profile [IAM사용자 이름]

 

그리고 Access Key ID, Secret Access Key를 차례대로 입력해준 뒤, 

 

작업중인 리전(ap-northeast-2)를 입력한다.

 

 

2. Producer 애플리케이션 작성

 

위에서 생성한 SQS에 비동기로 요청을 처리하는 Spring Application을 작성한다.

내가 만들고자하는 Application은 다음과 같다.

- 수강신청은 오전9시에 시작되며, 오전 9시~10시 사이에 발생하는 요청은 비동기로 처리한다.
   그 이후시간에는 동기로 처리한다.

- API 호출 시 SQS에 메시지를 전달한다.

 

2.1 애플리케이션 의존성 및 환경 설정

build.gradle

implementation platform("io.awspring.cloud:spring-cloud-aws-dependencies:3.0.2")
implementation 'io.awspring.cloud:spring-cloud-aws-starter-sqs'

 

application.properties

# ===============================
# = AWS SQS SETTINGS    =
# ===============================
sqs.queue.url=[sqs 엔드포인트]
spring.cloud.aws.sqs.enabled=true
cloud.aws.region.static=ap-northeast-2

 

2.2 SQS 관련 클래스 작성

SQSConfig.java

package com.example.moyeorak.config;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.SqsClient;

@Configuration
public class SQSConfig {
    
    @Value("${cloud.aws.region.static}")
    private String region;

    // SQS Client
    @Bean
    public SqsAsyncClient sqsAsyncClient() {
        return SqsAsyncClient.builder()
                .region(Region.of(region))
              // EC2 서버 환경에서 실행할 경우
                .credentialsProvider(DefaultCredentialsProvider.create()) 
              // 로컬 환경에서 실행할 경우 (IAM profile 설정)
                .credentialsProvider(ProfileCredentialsProvider.create("sqs-user"))
                .build();
    }

    @Bean
    public SqsClient sqsClient() {
        return SqsClient.builder()
                .region(Region.of(region))
              // EC2 서버 환경에서 실행할 경우
                .credentialsProvider(DefaultCredentialsProvider.create()) 
              // 로컬 환경에서 실행할 경우 (IAM profile 설정)
                .credentialsProvider(ProfileCredentialsProvider.create("sqs-user"))
                .build();
    }

}

 

위에는 비동기 방식을 위한 클라이언트고, 아래는 동기 방식을 위한 클라이언트다.

 

Spring Application에서는 AWS SQS에 접근하기 위해서는 클라이언트를 통해 접근해야한다.

 

이때, EC2 환경일 경우에는 메타데이터에서 IAM 권한을 가져와서 액세스 하기 때문에, 

"DefaultCrenentialsProvider.create()"로 접근하면되고 

 

로컬환경일 경우에는 aws-cli를 통해 생성한 IAM 사용자 profile을 통해 액세스 하기 때문에,
"ProfileCredentialsProvider.create("sqs-user")" 로 접근하게 된다.

 

사용하는 환경에따라 둘 중 하나를 주석처리하여 사용하면 된다.

 

AsyncEnrollmentService.java

import com.example.moyeorak.dto.EnrollmentMessage;
import com.example.moyeorak.dto.EnrollmentRequest;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.SendMessageRequest;

@Slf4j
@Service
@RequiredArgsConstructor
public class AsyncEnrollmentService {

    private final SqsAsyncClient sqsAsyncClient;
    private final ObjectMapper objectMapper;

    @Value("${sqs.queue.url}")
    private String queueUrl;

    public void sendEnrollment(EnrollmentRequest request, String email) {
        log.info("[ASYNC] 대관 신청 요청 큐에 전송 중... by {}", email);

        try {
            // 메시지 변환
            String payload = objectMapper.writeValueAsString(new EnrollmentMessage(request, email));

            // SQS 메시지 요청 생성
            SendMessageRequest messageRequest = SendMessageRequest.builder()
                    .queueUrl(queueUrl)
                    .messageBody(payload)
                    .build();

            // 메시지 전송
            sqsAsyncClient.sendMessage(messageRequest)
                    .thenAccept(response -> {
                        log.info("[ASYNC] SQS 전송 성공. 메시지 ID: {}", response.messageId());
                    })
                    .exceptionally(e -> {
                        log.error("[ASYNC] SQS 전송 실패", e);
                        return null;
                    });

        } catch (JsonProcessingException e) {
            log.error("메시지 변환 실패", e);
        }
    }
}

 

비동기 메시지를 생성하고, 전송하는 서비스 로직이다.

 

ProgramService.java

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.util.*;

@Slf4j
@Service
@RequiredArgsConstructor
@Transactional(readOnly = true)
public class ProgramService {
    private static final ZoneId KST = ZoneId.of("Asia/Seoul");
    
    public boolean isAsyncPeriod(Long programId) {
        Program program = programRepository.findById(programId)
                .orElseThrow(() -> new IllegalArgumentException("프로그램을 찾을 수 없습니다."));

        LocalDate registrationDate = program.getRegistrationStartDate();
        if (registrationDate == null) {
            log.warn("[isAsyncPeriod] registration_start_date가 null입니다. programId={}", programId);
            return false;
        }

        LocalDateTime now = LocalDateTime.now(KST);
        LocalDateTime start = registrationDate.atTime(9, 0);
        LocalDateTime end = registrationDate.atTime(10, 0);

        boolean result = now.isAfter(start) && now.isBefore(end);
        log.info("[isAsyncPeriod] programId={}, now={}, start={}, end={}, isAsync={}",
                programId, now, start, end, result);

        return result;
    }
}

 

나는 동기/비동기를 구분하는 기준을 "현재 시간"으로 잡았기 때문에, 

 

신청한 프로그램이 수강신청 시작 날짜의 오전 9-10인지 판단하여 동기/비동기 분기를 결정하도록 작성하였다.

 

이때 LocalDateTime.now()를 하면 UTC기준 시간으로 잡히기 때문에, 현재 한국 시간과 차이가 발생한다.

 

따라서 ZoneId를 KST로 설정하여, 현재 한국을 기준으로 시간을 확인하도록 작성했다.

 

EnrollmentMessage.java

@Getter
@AllArgsConstructor
public class EnrollmentMessage implements Serializable {
    private EnrollmentRequest request;
    private String email;

}

 

2.3 API 엔드포인트 작성

EnrollmentController.java

@Slf4j
@RestController
@RequestMapping("/api/enrollments")
@RequiredArgsConstructor
public class EnrollmentController {

    private final EnrollmentService enrollmentService;
    private final AsyncEnrollmentService asyncEnrollmentService;
    private final ProgramService programService;

    @PostMapping
    @PreAuthorize("hasRole('USER')")
    public ResponseEntity<?> enroll(
            @AuthenticationPrincipal CustomUserDetails userDetails,
            @Valid @RequestBody EnrollmentRequest request
    ) {
        String email = userDetails.getEmail();
        Long programId = request.getProgramId();
        log.info("[POST] 수강 신청 요청 - email: {}, programId: {}", email, programId);

        boolean async = programService.isAsyncPeriod(programId);
        if (async) {
            log.info("[ASYNC] 비동기 수강신청 처리 시작");
            asyncEnrollmentService.sendEnrollment(request, email);
            return ResponseEntity.accepted()
                    .body(new MessageResponse("비동기 수강신청 요청이 접수되었습니다."));
        } else {
            log.info("[SYNC] 동기 수강신청 처리 시작");
            return ResponseEntity.ok(enrollmentService.enrollByEmail(email, request));
        }
    }
}

 

/api/enrollments 경로를 통해 프론트로부터 요청을 받는 API 엔드포인트이다.

 

isAsyncPeriod()를 통해 비동기 요청인지 판별하여, 동기/비동기로 분기하여 요청을 처리하기 때문에, 

 

동기/비동기 모두 동일한 엔드포인트를 사용한다.

 

3. 테스트

 

ProgramService.java의 시간 설정 코드 중 end 값을 수정해서 테스트를 진행하는게 제일 간단한것 같다.

 

비동기 요청일 경우, 아래에 비동기 수강신청 요청이 접수되었다고 뜬다.

hani@Hani-MacBookPro 프로젝트 % curl -X POST \
'[요청을 보낼 주소]' \
-H 'Content-Type: application/json' \
-H 'accept: application/json' \
-H 'Authorization: Bearer [토큰값]
-d '{
"programId": 1,
"paidAmount": 10000
}'
{"message":"비동기 수강신청 요청이 접수되었습니다."}%