[실습] 이메일 발송을 처리할 Consumer 로직 짜기

JSCODE 박재성

2025-12-06

✅ 이메일 발송을 처리할 Consumer 로직 짜기

1. Kafka의 메시지를 가져와 담을 객체 만들기
UserSignedUpEvent

public class UserSignedUpEvent {
  private Long userId;
  private String email;
  private String name;

  // 역직렬화(String 형태의 카프카 메시지 -> Java 객체)시 필요함
  public UserSignedUpEvent() {
  }

  public UserSignedUpEvent(Long userId, String email, String name) {
    this.userId = userId;
    this.email = email;
    this.name = name;
  }

  public static UserSignedUpEvent fromJson(String json) {
    try {
      ObjectMapper objectMapper = new ObjectMapper();
      return objectMapper.readValue(json, UserSignedUpEvent.class);
    } catch (JsonProcessingException e) {
      throw new RuntimeException("JSON 파싱 실패");
    }
  }

  public Long getUserId() {
    return userId;
  }

  public String getEmail() {
    return email;
  }

  public String getName() {
    return name;
  }
}

 

2. Consumer 로직 작성하기
UserSignedUpEventConsumer

@Service
public class UserSignedUpEventConsumer {

	@KafkaListener(
      topics = "user.signed-up",
      groupId = "email-service",
      concurrency = "3"
  )
  @RetryableTopic(
      attempts = "5",
      backoff = @Backoff(delay = 1000, multiplier = 2),
      dltTopicSuffix = ".dlt"
  )
  public void consume(String message) throws InterruptedException {
    UserSignedUpEvent userSignedUpEvent = UserSignedUpEvent.fromJson(message);

    // 실제 이메일 발송 로직은 생략
    String receiverEmail = userSignedUpEvent.getEmail();
    String subject = userSignedUpEvent.getName() + "님, 회원 가입을 축하드립니다!";
    Thread.sleep(3000); // 이메일 발송에 3초 정도 시간이 걸리는 걸 가정
    System.out.println("이메일 발송 완료");
  }
}

위 로직에서 추가로 이메일 발송 로그를 DB에 저장하는 로직을 추가해야 한다. 이 로직을 추가하기 위해 필요한 엔티티와 레포지토리를 생성해주자.
 

3. 이메일 발송 로그를 남기기 위한 엔티티, 레포지토리 생성하기
EmailLog

@Entity
@Table(name = "email_logs")
public class EmailLog {
  @Id
  @GeneratedValue(strategy = GenerationType.IDENTITY)
  private Long id;

  private Long receiverUserId;

  private String receiverEmail;

  private String subject;

  public EmailLog() {
  }

  public EmailLog(Long receiverUserId, String receiverEmail, String subject) {
    this.receiverUserId = receiverUserId;
    this.receiverEmail = receiverEmail;
    this.subject = subject;
  }
  
  // getter 메서드
}

public interface EmailLogRepository extends JpaRepository<EmailLog, Long> {
}

4. Consumer 로직 보완하기
이메일 발송 로그를 DB에 저장하는 로직을 추가하자.
UserSignedUpEventConsumer

@Service
public class UserSignedUpEventConsumer {

  private EmailLogRepository emailLogRepository;

  public UserSignedUpEventConsumer(EmailLogRepository emailLogRepository) {
    this.emailLogRepository = emailLogRepository;
  }

  @KafkaListener(
      topics = "user.signed-up",
      groupId = "email-service",
      concurrency = "3"
  )
  @RetryableTopic(
      attempts = "5",
      backoff = @Backoff(delay = 1000, multiplier = 2),
      dltTopicSuffix = ".dlt"
  )
  public void consume(String message) throws InterruptedException {
    UserSignedUpEvent userSignedUpEvent = UserSignedUpEvent.fromJson(message);
    
    String receiverEmail = userSignedUpEvent.getEmail();
    String subject = userSignedUpEvent.getName() + "님, 회원 가입을 축하드립니다!";
    Thread.sleep(3000);
    System.out.println("이메일 발송 완료");

    EmailLog emailLog = new EmailLog(
        userSignedUpEvent.getUserId(),
        receiverEmail,
        subject
    );

    emailLogRepository.save(emailLog);
  }
}

 

5. DLT로 빠지는 메시지 처리하는 로직 추가하기
UserSignedUpEventDltConsumer

@Service
public class UserSignedUpEventDltConsumer {
  @KafkaListener(
      topics = "user.signed-up.dlt",
      groupId = "email-service"
  )
  public void consume(String message) {
    // 실제 로직은 생략
    System.out.println("로그 시스템에 전송 : " + message);
    System.out.println("Slack에 알림 발송");
  }
}

✅ 중간 체크