[실습] Spring Boot에서 하나의 컨슈머로 메시지 병렬적으로 처리하기

JSCODE 박재성

2025-12-06

✅ Spring Boot로 하나의 컨슈머로 메시지 병렬적으로 처리하기

1. 하나의 컨슈머 서버만 실행시키기
이전 실습으로 인해 컨슈머 서버를 두 대 띄워뒀다면 한 대를 종료시키자.



 

2. 컨슈머 서버의 코드 수정하기
EmailSendConsumer

@Service
public class EmailSendConsumer {

  @KafkaListener(
      topics = "email.send",
      groupId = "email-send-group",
      concurrency = "3" // 멀티 쓰레드를 활용해 병렬적으로 처리할 파티션의 개수
  )
  @RetryableTopic(
      // 총 시도 횟수 (최초 시도 1회 + 재시도 4회)
      attempts = "5",

      // 1 -> 2 -> 4 -> 8 순으로 재시도 시간이 증가한다.
      backoff = @Backoff(delay = 1000, multiplier = 2),

      // DLT 토픽 이름에 붙일 접미사
      dltTopicSuffix = ".dlt"
  )
  public void consume(String message) {
    System.out.println("Kafka로부터 받아온 메시지: " + message);

    EmailSendMessage emailSendMessage = EmailSendMessage.fromJson(message);

    // 잘못된 이메일 주소일 경우 실패 가정
    if (emailSendMessage.getTo().equals("fail@naver.com")) {
      System.out.println("잘못된 이메일 주소로 인해 발송 실패");
      throw new RuntimeException("잘못된 이메일 주소로 인해 발송 실패");
    }

    // ... 실제 이메일 발송 로직은 생략 ...
    try {
      Thread.sleep(10000); // 이메일 발송을 하는 데 10초가 걸린다고 가정
    } catch (InterruptedException e) {
      throw new RuntimeException("이메일 발송 실패");
    }

    System.out.println("이메일 발송 완료");
  }
}

 

3. 컨슈머 서버 재실행시키기



 

4. API 요청 보내면서 컨슈머 서버 로그 확인하기
API 요청을 연속으로 3번 보낸 뒤에 컨슈머 서버의 로그를 확인해보자. 로그를 보면 3개의 메시지를 병렬적으로 처리하는 걸 확인할 수 있다. Spring Boot 서버에서 총 3개의 쓰레드가 파티션을 하나씩 담당해서 처리하는 방식으로 작동하고 있기 때문이다.



 
그럼 이번에는 API 요청을 연속으로 5번 보낸 뒤에 컨슈머 서버의 로그를 확인해보자. 3개의 메시지까지는 바로 컨슈머 서버가 받아들이지만, 그 다음 2개의 메시지는 각 파티션의 이전 메시지가 처리가 되는 대로 받아들인다. 그 이유는 해당 토픽의 파티션이 3개이기 때문에 여러 쓰레드가 병렬적으로 처리할 수 있는 메시지의 개수가 최대 3개일 수 밖에 없기 때문이다.

✅ 그림으로 이해하기