[개발일지] spring batch를 통한 대량 이메일 발송 기능 개발 - 2 (AWS SES로의 마이그레이션, 적절한 스레드 개수와 청크 사이즈 계산, 쿼리 최적화, 스레드 동적 생성)

 

https://2junbeom.tistory.com/204

이전 포스트

[개발일지] spring batch를 통한 대량 이메일 발송 기능 개발 - 1 (멀티 스레딩 동시성 문제)

 

SES 선택 이유

대량 메일 발송 서비스에서 가장 유명한 서비스는 두 개가 있었다.

 

1. AWS SES

2. SendGrid

 

두 개의 장단점을 비교해보자

 

• SES
  • 장점
    1. 비용 저렴 : 1000건 당 $0.1
    2. 기존 배포 및 서비스를 모두 AWS에서 사용하고 있었기에 통합이 쉽고
       로그 관리에 용이함
  • 단점
    1. 저수준의 UI
    2. 이메일 템플릿은 직접 구현해야한다.
• SendGrid
  • 장점
    1. UI가 뛰어나다
    2. 이메일 템플릿을 제공한다.
  • 단점
    1. 비용이 비싸다: 1000건당 $1 ~ $3

우리 서비스는 AWS 인프라를 기반으로 구축되어 있고 비용을 절감해야하기 때문에 AWS SES를 선택했다.

 

SES 도메인 및 이메일 등록

ses에서 기존에 rotue 53에 등록해둔 도메인과

발송 이메일을 등록했다. 이 부분은 간단하니 패스

 

초기 이메일 발송 제한은 하루 200건 밖에 되지 않는다.

 

이 발송 제한을 해제하기 위해선 AWS에 발신 한도 증설 요청을 보내야한다.

 

Subject: Additional Information Regarding Our Email Practices

Hello,

Thank you for your response.

Our service sends email notifications exclusively to users who have explicitly opted in to receive updates. Specifically, we send announcements twice a year to inform users about the opening of a new application/support cycle.

In addition, our system handles routine transactional emails such as authentication code delivery and welcome messages during the signup process.

Here are additional details about our email practices:

Frequency:

Transactional Emails: Approximately 40–50 emails per day are sent for user authentication codes and signup confirmations.

Promotional Emails: Sent only twice a year (every 6 months) for recruitment announcements.

 

대충 뭐 이런 느낌으로 빈도, 발송 내용, 구독자에게만 보내는 건지 구독 해지는 어떻게 하는지 등등을 적어서 보내면 된다.

 

3일 정도 기다린 후에, 하루 5만 건, 초당 14개의 메일 전송이 가능하다는 메일을 받았다!

 

이제 코드를 작성해보자

 

기존 사용하던 AWS BACKEND IAM USER에 SESFULLACCESS 권한을 부여해줬고,

기존에 사용하던 액세스 키와 시크릿 키를 그대로 유지했다.

 

@Configuration
public class SESMailConfig {

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

    @Bean
    public AmazonSimpleEmailService amazonSimpleEmailService() {
        final BasicAWSCredentials basicAWSCredentials = new BasicAWSCredentials(accessKey, secretKey);
        final AWSStaticCredentialsProvider awsStaticCredentialsProvider = new AWSStaticCredentialsProvider(
                basicAWSCredentials);

        return AmazonSimpleEmailServiceClientBuilder.standard()
                .withCredentials(awsStaticCredentialsProvider)
                .withRegion("ap-northeast-2")
                .build();
    }
}

 

다음과 같이 설정해주고

 

메일 작성 코드를 만들어줬다.

@Slf4j
@Component
@RequiredArgsConstructor
public class SESMailUtil {

    private final AmazonSimpleEmailService amazonSimpleEmailService;

    @Value("${spring.mail.username}")
    private String mail;

    public SendRawEmailRequest getSendRawEmailRequest(String title, String content, String receiver, String html)
            throws MessagingException, IOException {

        Session session = Session.getDefaultInstance(getProperties());
        MimeMessage message = new MimeMessage(session);

        // Define mail title
        message.setSubject(title);

        // Define mail Sender
        message.setFrom(new InternetAddress(mail));

        message.setHeader("X-SES-CONFIGURATION-SET", "my-first-configuration-set");

        // Define mail Receiver
        message.setRecipients(Message.RecipientType.TO, InternetAddress.parse(receiver));

        // Create a multipart/alternative child container.
        MimeMultipart msg_body = new MimeMultipart("alternative");

        // Create a wrapper for the HTML and text parts.
        MimeBodyPart wrap = new MimeBodyPart();

        // Define the text part.
        MimeBodyPart textPart = new MimeBodyPart();
        textPart.setContent(content, "text/plain; charset=UTF-8");

        // Define the HTML part.
        MimeBodyPart htmlPart = new MimeBodyPart();
        htmlPart.setContent(html, "text/html; charset=UTF-8");

        // Add the text and HTML parts to the child container.
        msg_body.addBodyPart(textPart);
        msg_body.addBodyPart(htmlPart);

        // Add the child container to the wrapper object.
        wrap.setContent(msg_body);

        // Create a multipart/mixed parent container.
        MimeMultipart msg = new MimeMultipart("mixed");

        // Add the parent container to the message.
        message.setContent(msg);

        // Add the multipart/alternative part to the message.
        msg.addBodyPart(wrap);

        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        message.writeTo(outputStream);
        RawMessage rawMessage = new RawMessage(ByteBuffer.wrap(outputStream.toByteArray()));
        return new SendRawEmailRequest(rawMessage);

    }

    private Properties getProperties() {
        Properties props = System.getProperties();
        props.put("mail.transport.protocol", "smtp");
        props.put("mail.smtp.port", 587);
        props.put("mail.smtp.starttls.enable", "true");
        props.put("mail.smtp.auth", "true");
        return props;
    }

}

 

이제 아래의 방식으로 SES Service SDK를 통해 메일을 보낼 수 있다.

private final AmazonSimpleEmailService emailService;


try {
            SendRawEmailRequest request = sesMailUtil.getSendRawEmailRequest(
                    "TAVE 4기 모집 안내",
                    "TAVE 4기 모집이 시작되었습니다. 자세한 내용은 이메일 본문을 확인해주세요.",
                    recipient,
                    getApplyNotificationTemplate()
            );
         	emailService.sendRawEmail(request);
}

 

지수 백오프 도입

SES 초당 최대 전송 가능 수는 14개다.

멀티 스레딩으로 병렬 처리를 하면 초당 14개를 넘기는 경우가 발생할 수 있다.

 

이 경우 메일 발송에 실패하게 되는데, 우리는 단 한 건의 메일도 실패하지 않는 것을 목표로 하고 있다.

 

우선 적절한 스레드 갯수를 구해보자.

https://en.wikipedia.org/wiki/Little%27s_law

ThreadPoolSize = (Max TPS × Processing Time per Request) / (1 second)

 

여기서 Max TPS는 초당 14건 즉 14이고

메일 한 건당 전송 속도는 이제 측정해보도록 하자.

 

메일 한 건당 전송 속도

 

scouter로 바로 측정했고 10건의 평균 Elapse를 구했다.

 

평균 elapse는 182ms

 

그럼 TPS가 14인 상황에서 elapse가 182ms니까 

2.548개가 적절한 스레드 수로 나온다.

 

그럼 스레드 수는 3개로 설정하고 돌리면 되겠다!

 

다만 스레드가 3개일 때 초당 16건까지 메일을 보낼 수도 있으니, 지수 백오프를 도입해서

초당 14회가 넘는 요청에 대해선 지수적으로 대기 후 재시도를 하도록 로직을 구성했다.

우린 이를 통해 최대 효율료, 실패 없이 많은 이메일을 보낼 수 있을 것이다.

 

scouter 킨 김에 적절한 chunk 사이즈까지 구해보자

 

적절한 chunk Size 구하기

우리가 사용하는 서버는 aws t3.micro다.

 

메모리 사이즈는 1GB이고, 힙 메모리는 512mb로 설정돼 있다.

 

아무리 새벽 세 시에 돌린다지만 힙 메모리를 다 점유하게 만들 수는 없으니 적절한 청크 사이즈를 통해,

메모리를 관리해야한다.

 

GC 여유 공간을 위해 100mb를 남겨두고

Spring + hibrenate 등등을 위해 약 100mb를 더 빼자

 

여유 공간은 약 300mb, 새벽 3시이지만 다른 요청이 들어올 수도 있으니 여유 50mb 정도 남겨두고

 

힙 메모리를 250mb까지 사용하는 걸 목표로 잡고 적절한 청크 사이즈를 정해보자.

사용하는 라이브러리가 많아서 그런지 아무 요청이 없는 상태에서 heap used가 150mb 정도로 측정된다.

 

힙 메모리 사용 목표치를 200mb로 설정하고 다시 수행,

 

스레드 개수가 적으므로 청크 사이즈를 크게 잡고 시작할 수 있을 것 같다.

 

(청크 사이즈는 클 수록 좋다. - > 스레드 전환 비용, 디비 커넥션 횟수, 커밋 등 I/O작업 횟수 감소 등등 I/O 작업을 줄여야 하니 청크를 크게 잡는 게 이득이라 판단했다.)

 

일단 시작 청크 사이즈를 500으로 잡아보자.

 

@Bean
public Step emailNotificationStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
    return new StepBuilder("emailNotificationStep", jobRepository)
            .<EmailNotification, EmailNotification>chunk(500, transactionManager)
            .reader(emailReader)
            .writer(emailWriter)
            .build();
}

@Bean
public Partitioner idRangePartitioner() {
    return new IdRangePartitioner(emailNotificationRepository);
}

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(3);
    executor.setMaxPoolSize(3);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("batch-thread-");
    executor.initialize();
    return executor;
}

 

자 청크 사이즈는 500이고, 스레드 개수는 아까 위에서 계산한대로 3개로 설정해주었다.

 

최대 253mb 약 100mb 정도를 차지한다.

 

스레드가 3개이므로 한 번에 1500 청크가 메모리에 올라가는데, 

1500 청크에 100mb이므로, 200mb까지 사용가능하다 하면

3000 청크까지 가능하다. 3000 청크면 스레드 별로 1000 chunk를 할당해주면 되니,

 

청크의 개수는 1000개로 할당해주자.

 

Jpa Item Reader와 파티셔닝 선택 이유

• 우선 사용 가능한 Reader의 종류는 다음과 같다.
  • JpaPagingItemReader - JPA기반 페이징 조회,  스레드 세이프하다.
  • JdbcPagingItemReader - JDBC기반 페이징 조회, 스레드 세이프
  • ListItemReader - 메모리상의 List를 순서대로 읽음, 스레드 세이프 하지 않으며 필요한 데이터를 모두 가져와 메모리에 올려두고 사용한다. 데이터가 많다면 Out Of Meory의 원인이 될 수 있다.

우리 서비스는 jpa를 사용하고 멀티 스레딩으로 작업할 것이니 jpaPagingItemReader를 사용해주자.

 

그럼 이제 병렬 처리를 어떻게 할 것인가를 선택해야한다.

 

1. 멀티 스레드 스텝 - 스텝 내의 각 청크를 여러 스레드로 병렬 실행하는 방법 - reader, writer가 스레드 세이프 하지 않다면
   동시성 문제가 발생할 수 있다.
2. 파티셔닝 - 전체 사이즈를 파티션으로 나눠 각 파티션 별로 스레드가 할당되어 독립적으로 작업을 수행하는 방식
   파티션 별로 작업을 수행하기 때문에 reader, writer가 스레드 세이프 한지 안 한지 고려할 필요 없다.
3. 병렬 스텝 - 여러 개의 스텝을 병렬로 수행하는 방식 - 스레드 세이프하다.

우선 ItemWriter를 직접 구현할 것이기 때문에 동시성문제 고려를 가장 적게 하고 싶었다.

파티셔닝과 병렬 스텝 둘 중 고민을 많이 했는데, 둘 사이의 차이가 거의 없어 
STATUS가 PENDING 상태인 이메일을 ID기준으로 파티셔닝 하는 로직이 적합할 것 같아

파티셔닝 방식을 선택했다.

 

좀 더 자세히 설명하면 파티셔닝 방식은 전체 데이터 범위를 설정한 그리드 사이즈로 나누고

자신의 파티션을 청크 사이즈 별로 독립적으로 처리하는 방식이다.

테스트 중 쿼리 최적화

나는 기존에 JpaWriter를 사용하고 있었다.

 

그러나 JpaWriter는 내부적으로 entitymanager.maerge() 메서드를 사용하는데

 

merge 메서드는 select와 insert 두 번의 쿼리가 모든 영속성 컨텍스트 별로 나간다.

 

난 이점이 굉장히 마음에 안 들었고 나중에 QueryDsl로 Reader를 구현할 일이 생길 수도 있으니

그냥 직접 writer를 구현하기로 했다. 어차피 파티셔닝 써서 동시성 문제 많이 고려 안 해두 됨 ㅎㅎ

 

그렇게 writer를 구현하던 중

2000개의 이메일 테스트 중 쿼리가 2000번이 나가는 아주 심각한 문제점 발견

 

이유 분석

 

writer에서 saveAll 호출이 원인이었다.

 

나는 saveAll이 벌크 연산인줄 알았으나 내부를 살펴보니 벌크 연산이 아니었다 ㅠㅠㅠ

 

따로 벌크 연산 쿼리를 작성해주자.

@Modifying(clearAutomatically = true)
@Query("UPDATE EmailNotification e SET e.status = :status, e.retryCount = e.retryCount + 1, e.updatedAt = CURRENT_TIMESTAMP WHERE e.id IN :ids")
void bulkUpdateStatus(@Param("status") EmailStatus status, @Param("ids") List<Long> ids);

 

이후 writer를 다음과 같이 변경

return items -> {
    List<Long> successIds = new ArrayList<>();
    List<EmailNotification> failedItems = new ArrayList<>();

    for (EmailNotification item : items) {
        try {
            retryTemplate.execute(context -> {
                sesMailService.sendApplyNotification(item.getEmail());
                log.info("메일 전송 성공: {}", item.getEmail());
                successIds.add(item.getId());
                return null;
            }, context -> {
                item.changeStatus(EmailStatus.FAILED);
                failedItems.add(item);
                log.error("DLQ 처리 - {}: {}", item.getEmail(),
                        Objects.requireNonNull(context.getLastThrowable()).getMessage());
                return null;
            });
        } catch (Exception e) {
            item.changeStatus(EmailStatus.FAILED);
            failedItems.add(item);
            log.error("Unexpected failure: {}", item.getEmail(), e);
        }
    }

    // 1. 성공 건: 벌크 업데이트
    if (!successIds.isEmpty()) {
        emailNotificationRepository.bulkUpdateStatus(EmailStatus.SUCCESS, successIds);
    }

    // 2. 실패 건: 개별 필드 변경사항 포함하여 saveAll
    if (!failedItems.isEmpty()) {
        emailNotificationRepository.saveAll(failedItems);
        emailNotificationRepository.flush();
    }
};

 

실패 건에 대해서는 saveAll() 형태를 유지한다. (발생 확률이 거의 없기 때문)

 

쿼리 2000번 -> 6번으로 무사히 구현완료 !

 

아무튼 심각한 쿼리 오류와 멀티 스레딩 문제 해결!

 

쿼리 몇 번 나가는지 확인 안 했으면 진짜 큰일날 뻔했다 ㅎㅎ

 

다음 포스트에서는 분산락을 활용하여 Shecduler 어노테이션을 도입한 내용에 대해 작성하겠다.

 

https://2junbeom.tistory.com/206?category=1256638

[개발일지] spring batch를 통한 대량 이메일 발송 기능 개발 - 3 (redis Shed lock, @Scheduled, 분산락)