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

spring batch 도입

현재 진행 중인 프로젝트에서 이메일을 대량으로 발송하는 기능을 개발해야했다.

 

대량 메일 발송 기능을 위해 spring batch 기능을 도입했다.

 

spring batch 도입 이유는 다음과 같다.

• 메모리 최적화
  • 한 번에 모든 객체를 메모리에 올려서 처리하는 게 아니라 chunk 단위로 메모리에 올려 처리하기 때문에,
    메모리 최적화에 용이하다.
• 쉬운 재시도
  • 이메일 발송의 경우 초당 전송량이 제한돼 있기 때문에, 이메일 전송에 실패할 가능성이 높다.
    모든 회원에게 이메일을 실패없이 보내는 것을 목표로 하기 때문에, 재시도는 필수이고
    이러한 재시도는 spring batch에서 쉽게 구성할 수 있다.
• 병렬 처리 기능
  • 멀티 스레딩, 파티셔닝 등의 기능을 지원하여 이를 기반으로 원할한 병렬 처리가 가능하다.

이러한 이유들을 기반으로 spring batch를 도입하였다.

 

멀티 스레딩 동시성 이슈 발생

구체적인 스레드 수를 정하기 전, 테스트 목적으로 이메일 2000개를 전송하고 있었다.

 

그러나 2000개 중 500개 정도의 메일뿐이 전송되지 않은 것을 확인할 수 있었다.

 

이유는 다음과 같았다.

@Bean
public PlatformTransactionManager transactionManager(EntityManagerFactory entityManagerFactory) {
    return new JpaTransactionManager(entityManagerFactory);
}

 

배치 job에 transactionManager로 entityManager를 주입해줬는데

 

엔티티 매니저는 싱글 스레드 환경에 맞게 설계됐다는 것을 깜빡한 것이다.

 

• 엔티티 매니저는 더티 체킹, 자동 플러시 등의 작업을 위해, 각 request 별로 별도의 객체가 생성되며
  해당 객체 내부에서 프록시 객체들의 상태를 관리한다.
  여기서 프록시 객체의 상태가 변했다면 자동으로 더티 체킹을 통해 동기화를 진행하고
  이 작업을 위해서 싱글 스레드로 운영하는 것이다. (멀티 스레딩이라면 내부 객체에 여러 스레드가 접근해 프록시 객체의 상태가 계속 변하기 때문 한 마디로 의도한 대로 흘러가지 않는다.)
• 이 엔티티 매니저 하나에 여러 개의 스레드를 할당 했기 때문에 동시성 문제가 발생한 것이다.

문제가 발생한 부분을 살펴보자

@Configuration
public class EmailReaderConfig {

    @Bean
    public JpaPagingItemReader<EmailNotification> emailReader(EntityManagerFactory entityManagerFactory) {
        JpaPagingItemReader<EmailNotification> reader = new JpaPagingItemReader<>();
        reader.setEntityManagerFactory(entityManagerFactory);
        reader.setQueryString("SELECT e FROM EmailNotification e WHERE e.status = 'PENDING'");
        reader.setPageSize(50);
        reader.setSaveState(false);
        reader.setName("emailReader");
        return reader;
    }
}

 

이 코드에서 JpaPagingItemReader를 통해 페이징 수를 자동으로 증가시키며 다음 페이지를 읽어오고 배치 처리를 한다.

 

JpaPagingItemReader는 다음과 같은 상태를 유지한다.

• entityManager
• CurrentPage
• Query

위의 상태를 모든 스레드에서 공유하기 때문에 동시성 문제가 발생하는 것이다.

 

currentPage를 공유하기 때문에 같은 페이지를 읽어올 수 있고,

하나의 엔티티 매니저를 공유하기 때문에 데이터베이스 업데이트 시 문제가 발생한다.

 

이 문제를 해결하기 위해 Partitioner + 각 파티션별 Reader 구조를 사용했다.

이전 구조는 한 개의 Reader를 사용했기 때문에 문제가 생겼고,

파티션 스텝은 각 스텝 별 각기 다른 Reader를 만들고, 각각 다른 엔티티 매니저를 할당하며

스텝별로 범위를 분할해서 읽어오기 때문에 동시성 문제를 해결할 수 있는 것이다.

 

@Configuration
public class EmailReaderConfig {

    @Bean
    @StepScope // step scope로 런타임 시 동적으로 변수 값 할당
    public JpaPagingItemReader<EmailNotification> emailReader(
            @Value("#{stepExecutionContext['startId']}") Long startId,
            @Value("#{stepExecutionContext['endId']}") Long endId,
            EntityManagerFactory entityManagerFactory
    ) {
        JpaPagingItemReader<EmailNotification> reader = new JpaPagingItemReader<>();
        reader.setEntityManagerFactory(entityManagerFactory);
        reader.setQueryString(
                "SELECT e FROM EmailNotification e WHERE e.status = 'PENDING' AND e.id BETWEEN :startId AND :endId");
        reader.setParameterValues(Map.of("startId", startId, "endId", endId));
        reader.setPageSize(50);
        reader.setSaveState(false);
        reader.setName("emailReader");
        return reader;
    }
}

 

수정된 코드는 다음과 같다.

StepScope 어노테이션을 통해 Reader 생성시 마다 동적으로 가져올 페이지를 할당한다.

 

이후 파티션 별로 쿼리를 날려 데이터를 가져온다.

 

파티션은 다음과 같이 생성했다.

public class IdRangePartitioner implements Partitioner {

    private final EmailNotificationRepository repository;

    public IdRangePartitioner(EmailNotificationRepository repository) {
        this.repository = repository;
    }

    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        Long minId = repository.findMinId();
        Long maxId = repository.findMaxId();

        Map<String, ExecutionContext> result = new HashMap<>();

        if (minId == null || maxId == null) {
            return result;
        }

        long targetSize = (maxId - minId) / gridSize + 1;
        long start = minId;
        long end = start + targetSize - 1;

        for (int i = 0; i < gridSize; i++) {
            ExecutionContext context = new ExecutionContext();
            context.putLong("startId", start);
            context.putLong("endId", Math.min(end, maxId));
            result.put("partition" + i, context);
            start += targetSize;
            end += targetSize;
        }

        return result;
    }
}

 

전체 이메일의 크기를 가져오고 그것을 파티션 사이즈 별로 분할해서 실행 context에 저장시켜둔다

이후 실행 context에서 StepScope를 통해 동적으로 할당 받아와 Reader에서 읽어온다.

 

이를 통해 정상적으로 이메일을 발송할 수 있는 환경을 만들었다.

 

그러나 여기서 발생한 또 하나의 문제,,

 

naver smtp는 1시간 동안 최대 ???회, 1회에 최대 100명에게 메일을 보낼 수 있다.

https://help.naver.com/service/30029/contents/21159?lang=ko&osType=COMMONOS

메일 발송 제한 및 한 번에 보낼 수 있는 수신인 수 : 메일 고객센터

구체적인 수치는 적혀져 있지 않지만, 내가 테스트를 진행할 땐 100명에서 더 이상 보내지지 않았다.

 

(사실 이 문제는 너무 당연한 거라 이미 SES로 마이그레이션을 생각하고 있었다.)

 

다음 포스트에서 SES로의 마이그레이션을 다루겠다.

 

 

다음 포스트

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

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