kafka를 활용한 SMTP 메일 전송 (with. SpringBoot, redis)

1. 메일 전송 비동기 플로우

스케줄러 -> redis 캐시 -> kafka 발행 -> Consumer -> STMP 메일 전송 -> Mailtrap(https://mailtrap.io/home) 저장


시퀀스 다이어그램

2. 기술 스택

• 언어/프레임워크: Java 17, Spring Boot 3.4
• DB: MySQL (JPA/Hibernate)
• 캐시: Redis 
• 메시징: Apache Kafka
• 스케줄링: Spring Scheduling (@EnableScheduling, @Scheduled)
• 메일: Spring Boot Starter Mail + Mailtrap
• 컨테이너: Docker Compose (Zookeeper, Kafka, Redis)

3. 프로젝트 설정 (build.gradle)

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-data-redis'
    implementation 'org.springframework.kafka:spring-kafka'
    implementation 'org.springframework.boot:spring-boot-starter-mail'
}

 

4. Docker Compose로 redis, kafka, zookeper 띄우기

docker-compose.yml

cd docker-compose.yml 위치이동
docker compose down
docker compose up -d

5. application.yml 설정

spring:
  redis:
    host: localhost
    port: 6379

  kafka:
    bootstrap-servers: localhost:9092

  mail:
    host: smtp.mailtrap.io
    port: 2525
    username: username # mailtrap에서 message 참고
    password: password 

app:
  kafka:
    topic:
      email: email-topic

6. kafkaConfig & 메시징 컴포넌트 정리

6.1 kafkaConfig - 직렬화/역직렬화 설정

@Configuration
@EnableKafka
public class KafkaConfig {
    @Value("${spring.kafka.bootstrap-servers:localhost:9092}")
    private String bootstrapServers;

    @Value("${app.kafka.topic.email:email-topic}")
    private String topicName;

    // 1) ProducerFactory: 메시지 프로듀서를 생성하는 팩토리
    @Bean
    public ProducerFactory<String, EmailPayload> producerFactory() {
        Map<String,Object> props = new HashMap<>();
        // Kafka 브로커 주소
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        // 메시지 키/값 직렬화 방식
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(props);
    }

    // 2) KafkaTemplate: 실제로 메시지를 발행(send)할 때 사용하는 핵심 API
    @Bean
    public KafkaTemplate<String, EmailPayload> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    // 3) ConsumerFactory: 컨슈머 인스턴스를 생성하는 팩토리
    @Bean
    public ConsumerFactory<String, EmailPayload> consumerFactory() {
        JsonDeserializer<EmailPayload> deserializer = new JsonDeserializer<>(EmailPayload.class);
        // 패키지 신뢰 설정 (역직렬화 시 보안)
        deserializer.addTrustedPackages("com.quoteline.quote_server.notification.domain");

        Map<String,Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "quote-mail");
        // 키/값 역직렬화 클래스
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

        return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), deserializer);
    }

    // 4) ListenerContainerFactory: @KafkaListener를 동작시키는 컨테이너 설정
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, EmailPayload>
           kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, EmailPayload> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 옵션: 에러 핸들러, 동시 컨슈머 수 등 추가 설정 가능
        return factory;
    }

    // 5) NewTopic: 애플리케이션 시작 시 토픽이 없으면 자동 생성
    @Bean
    public NewTopic emailTopic() {
        // 파티션 개수:1, 레플리카 팩터:1
        return new NewTopic(topicName, 1, (short)1);
    }
}

producerFactory는 kafka 프로듀서를 구성하는 기본 속성입니다. KafkaTemplate는 메시지를 지정된 토픽으로 보내는 API이고 consummerFactory는 kafka 컨슈머를 구성하는 기본속성을 세팅합니다. kafkaListnerContainerFactory는 @KafkaLinstner가 동작할 컨슈머 컨테이너 팩토리로 병렬 리스닝 및 에러 핸들링을 설정합니다.
emailTopic은 애플리케이션 구동 시 자동으로 email-topic을 생성합니다.


6.2 EmailPayload - 메시지 페이로드 정의

@NoArgsConstructor
@AllArgsConstructor
@Getter
public class EmailPayload {
    @Email @NotBlank
    private String email;

    @NotBlank
    private String quote;
}

• DTO 역할: 프로듀서가 발행하고, 컨슈머가 수신할 때 사용하는 메시지 모델
• Validation: @Email, @NotBlank (컨트롤러 바인딩 혹은 수동 검증 시에만 동작)

 
6.3 NotificationProducer.java – Kafka 메시지 발행

@Service
@RequiredArgsConstructor
public class NotificationProducer {
    private final KafkaTemplate<String, EmailPayload> kafkaTemplate;

    @Value("${app.kafka.topic.email}")
    private String topic;

    public void send(EmailPayload payload) {
        // key: 수신자 이메일, value: EmailPayload
        kafkaTemplate.send(topic, payload.getEmail(), payload);
    }
}

• send() 호출 시 비동기로 브로커에 메시지 전송
• partition key 로 이메일을 사용하면, 같은 이메일 간 메시지는 동일 파티션에 정렬 보장

 
6.4 NotificationConsumer.java – Kafka 메시지 수신 & 메일 전송

@Service
@Slf4j
@RequiredArgsConstructor
public class NotificationConsumer {
    private final JavaMailSender mailSender;

    @KafkaListener(
      topics = "${app.kafka.topic.email}",
      groupId = "quote-mail"
    )
    public void consume(EmailPayload payload) {
        // 1) 메일 작성
        SimpleMailMessage msg = new SimpleMailMessage();
        msg.setFrom("no-reply@quoteline.com");
        msg.setTo(payload.getEmail());
        msg.setSubject("오늘의 명언");
        msg.setText(payload.getQuote());

        // 2) SMTP 전송
        mailSender.send(msg);
        log.info("Sent email to {} › {}", payload.getEmail(), payload.getQuote());
    }
}

• @KafkaListener가 메시지를 폴링 → consume() 호출
• JavaMailSender로 동기 메일 전송
• 예외 발생 시 기본 DefaultErrorHandler가 재시도 로직 수행

7. RedisConfig & 캐시 컴포넌트

@Configuration
public class RedisConfig {
    // ① LettuceConnectionFactory: 기본 싱글 노드 Redis 연결 팩토리
    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        return new LettuceConnectionFactory("localhost", 6379);
    }

    // ② RedisTemplate<String,String>: 문자열 키/값 전용 템플릿
    @Bean
    @Primary
    public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory cf) {
        RedisTemplate<String, String> tpl = new RedisTemplate<>();
        tpl.setConnectionFactory(cf);
        // 키·값을 모두 StringSerializer로 직렬화
        StringRedisSerializer s = new StringRedisSerializer();
        tpl.setKeySerializer(s);
        tpl.setValueSerializer(s);
        tpl.afterPropertiesSet();
        return tpl;
    }
}

• LettuceConnectionFactory
  • localhost:6379에 Redis 접속.
• RedisTemplate
  • opsForValue()로 단순 문자열 캐시: get, set(key,val,TTL) 호출 가능.
  • @Primary 지정해, Spring이 StringRedisTemplate 대신 여기 빈을 우선 주입하도록 함

 

8. 스케줄러: DailyNotificationScheduler

@Service
@RequiredArgsConstructor
public class DailyNotificationScheduler {
    private final SubscriptionRepository repo;
    private final QuoteService quoteService;
    private final NotificationProducer producer;

    // 매일 오전 9시 실행
    @Scheduled(cron = "0 0 9 * * *")
    public void sendMorningQuotes() {
        send(TimeSegment.AM);
    }

    // 매일 오후 6시 실행
    @Scheduled(cron = "0 0 18 * * *")
    public void sendEveningQuotes() {
        send(TimeSegment.PM);
    }

    private void send(TimeSegment seg) {
        // 1) 해당 시간대 구독자 조회
        List<Subscription> subs = repo.findByTimeSegment(seg);
        // 2) 캐시 혹은 신규 명언 발급
        String quote = quoteService.getQuote();
        // 3) Kafka 발행
        subs.forEach(s -> {
            EmailPayload payload = new EmailPayload(s.getEmail(), quote);
            producer.send(payload);
        });
    }
}

• @EnableScheduling 어노테이션이 QuoteServerApplication에 선언되어 있어야 작동
• cron 표현식으로 매일 오전·오후 스케줄링
• DB → 캐시 → 메시징 → 메일 전송의 트리거 역할

추후, 오전 시간대와 오후 시간대를 늘려서 사용자가 다양한 시간에 알림을 받을 수 있도록 확장할 예정입니다.

9. 전체 흐름 요약

• 스케줄러가 매일 지정된 시각에 구독자를 조회
• Docker Compose로 Zookeeper·Kafka·Redis 로컬 환경 일괄 구성
• QuoteService가 Redis 캐시로 하루 1회 명언 발급
• NotificationProducer가 Kafka 토픽에 EmailPayload 발행
• NotificationConsumer가 Kafka에서 메시지 수신 후 JavaMailSender로 SMTP 전송
• Mailtrap 웹 UI에서 실제 메일 수신 확인
• Docker Compose로 Zookeeper·Kafka·Redis 로컬 환경 일괄 구성