[실습] Spring Boot로 Kafka에서 메시지 조회하기 (Consumer)

JSCODE 박재성

2025-12-06

✅ Spring Boot로 Kafka에서 메시지 조회하기 (Consumer)

1. Spring Boot 프로젝트 셋팅
start.spring.io
https://start.spring.io/



• Artifact와 Name을 email-send-consumer라고 지어주자.
• Package name을 emailsendconsumer라고 지어주자.
• Java 21 버전을 선택하자. 아래 과정을 Java 21 버전을 기준으로 진행할 예정이다.
• Dependencies는 Spring Boot DevTools, Spring Web, Spring for Apache Kafka를 선택해라.
 

3. application.yml에 Kafka 연결을 위한 정보 작성하기

이 프로젝트에서는 application.properties를 지우고 application.yml을 생성했다.

application.yml

server:
  port: 0 # 사용 가능한 랜덤 포트를 찾아서 서버를 실행 (Producer 서버와의 포트 충돌을 방지)

spring:
  kafka:
    # Kafka 서버 주소 (EC2에 카프카를 설치했기 때문에 EC2 주소를 입력해야 한다.)
    bootstrap-servers: 15.164.96.71:9092
    consumer:
      # 메시지의 key 역직렬화 방식 : Kafka에서 받아온 메시지를 String으로 변환
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 메시지의 value 역직렬화 방식 : Kafka에서 받아온 메시지를 String으로 변환
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      
      # 컨슈머 그룹이 미리 안 만들어져있는 경우에, 컨슈머 그룹을 직접 생성해서 메시지를 처음부터 읽음.
      # 만약 컨슈머 그룹이 이미 만들어져있다면, 해당 컨슈머 그룹이 읽었던 메시지부터 읽음.
      
      # 이 옵션을 주지 않으면 컨슈머 그룹을 직접 생성해서 메시지를 읽을 때, 
      # 기존에 쌓여있던 메시지를 읽지 않고 컨슈머 그룹이 생성된 이후에 들어온 메시지부터 읽어버린다.
      # 그럼 컨슈머 그룹이 생성되기 전에 쌓여있던 메시지들이 처리되지 않고 누락돼버린다.
      auto-offset-reset: earliest
      

 

3. Kafka의 메시지를 가져와 담을 객체 만들기
EmailSendMessage (Kafka로부터 전달받는 메시지의 정보를 담을 객체)

public class EmailSendMessage {
  private String from;         // 발신자 이메일
  private String to;           // 수신자 이메일
  private String subject;      // 이메일 제목
  private String body;         // 이메일 본문
  
  // 역직렬화(String 형태의 카프카 메시지 -> Java 객체)시 필요함
  public EmailSendMessage() {
  }

  public EmailSendMessage(String from, String to, String subject, String body) {
    this.from = from;
    this.to = to;
    this.subject = subject;
    this.body = body;
  }
  
  // Json 값을 EmailSendMessage로 역직렬화하는 메서드
  public static EmailSendMessage fromJson(String json) {
    try {
      ObjectMapper objectMapper = new ObjectMapper();
      return objectMapper.readValue(json, EmailSendMessage.class);
    } catch (JsonProcessingException e) {
      throw new RuntimeException("JSON 파싱 실패");
    }
  }

  public String getFrom() {
    return from;
  }

  public String getTo() {
    return to;
  }

  public String getSubject() {
    return subject;
  }

  public String getBody() {
    return body;
  }
}

 
EmailSendConsumer

@Service
public class EmailSendConsumer {

  @KafkaListener(
      topics = "email.send",
      groupId = "email-send-group" // 컨슈머 그룹 이름
  )
  public void consume(String message) {
    System.out.println("Kafka로부터 받아온 메시지: " + message);
    
    EmailSendMessage emailSendMessage = EmailSendMessage.fromJson(message);

    // ... 실제 이메일 발송 로직은 생략 ...

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

• 선언한 컨슈머 그룹이 없으면 컨슈머 그룹을 생성해서 메시지를 읽어들인다.
• 컨슈머 그룹을 활용해서 메시지를 읽기 때문에, 어디까지 메시지를 읽었는 지도 Kafka에 같이 저장된다. 이 덕분에 순차적으로 메시지를 읽어나가면서 처리하게 된다.

✅ 잘 작동하는 지 테스트해보기

1. Spring Boot 서버 실행시키기



Kafka의 email.send 토픽에 쌓여있던 메시지를 가져와 처리한 걸 확인할 수 있다.
 

2. API 요청을 통해 Kafka에 메시지를 추가로 넣어보기
Producer 역할을 하는 Spring Boot에 API 요청을 보내 Kafka에 실시간으로 메시지를 넣어보자.





Consumer 역할을 하는 Spring Boot 서버가 Kafka의 email.send 토픽에 쌓인 메시지를 실시간으로 처리하는 걸 확인할 수 있다.
 

3. CLI로 Consumer Group 확인해보기

# 컨슈머 그룹 세부 정보 조회하기
$ bin/kafka-consumer-groups.sh \
	--bootstrap-server localhost:9092 \
	--group email-send-group \
	--describe



CURRENT-OFFSET(다음에 읽을 메시지의 오프셋 번호)이 2이므로 0~1의 오프셋 메시지는 이미 읽었다는 뜻이다. 이 출력값을 보니 컨슈머 그룹이 정상적으로 잘 작동했다는 걸 확인할 수 있다.
 
 

✅ 그림 보면서 전체 흐름 정리하기