Skip to main content

Spring Boot & Kafka


Intro

1. 왜 그냥 API 호출로는 안 될까?​

프로젝트에서 사용자 10,000명에게 이메일을 보내는 기능을 만든 적이 있다. 처음에 짠 코드는 단순했다.

// 나쁜 예: 동기식 직접 호출
for (User user : users) {
    emailClient.send(user.getEmail()); // 여기서 외부 서버가 느려지면 내 서버도 같이 멈춤!
}

문제는 두 가지였다. 외부 이메일 서버가 느려지면 내 서버 스레드가 거기서 다 막혀버린다. 그리고 내 서버는 빠르게 쏠 수 있어도 받는 쪽이 처리 못하면 그쪽 서버가 터진다.

이 문제를 Kafka로 해결했다.

2. Apache Kafka란?​

Producer가 메시지를 Topic에 쌓아두면 Consumer가 자기 속도에 맞춰 꺼내가는 구조다. 두 서버가 직접 연결되지 않으니 한쪽이 느려져도 다른 쪽이 영향을 받지 않는다.

2.1. 핵심 용어 정리​

Kafka를 다루기 위해서는 다음 4가지 용어를 알아야 한다.

  • Producer (생산자): 데이터를 보내는 쪽이다. 예제에서는 이메일 발송 요청을 하는 API 서버를 의미한다.
  • Consumer (소비자): 데이터를 가져가서 처리하는 쪽이다. 이메일을 실제로 발송하는 서비스를 의미한다.
  • Topic (토픽): 데이터가 저장되는 주제 또는 폴더다. 예시는 email-send-job이다.
  • Offset (오프셋): 책갈피다. Consumer가 어디까지 읽었는지 표시하는 숫자다.

핵심 원리는 다음과 같다. Producer는 Consumer가 바쁘든 말든 상관하지 않고 Topic에 데이터를 밀어 넣는다. Consumer는 자기 속도에 맞춰서 Topic에서 데이터를 꺼내간다. 두 시스템의 속도 차이를 이런 방식으로 완충한다.

3.1. 인프라 구축 (Docker Compose)​

Docker를 통해 로컬환경에 Kafka를 설치해보았다.

# docker-compose.yml
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

터미널에서 docker-compose up -d를 입력하면 Kafka가 실행된다.

3.2. 의존성 추가 (build.gradle)​

spring-kafka 라이브러리를 추가한다.

dependencies {
    implementation 'org.springframework.kafka:spring-kafka'
}

3.3. 설정 (application.properties)​

Kafka 접속 정보와 Consumer 그룹 정보를 설정한다.

# Kafka 서버 주소
spring.kafka.bootstrap-servers=localhost:9092

# 데이터를 직렬화/역직렬화할 방식 (String 사용)
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

# Consumer 그룹 ID
spring.kafka.consumer.group-id=email-consumer-group

# 처음 실행할 때 데이터를 어디서부터 읽을 것인가?
spring.kafka.consumer.auto-offset-reset=earliest

3.4. Producer 구현 (보내는 쪽)​

KafkaTemplate을 주입받아 사용하면 매우 간단하다.

@Service
@RequiredArgsConstructor
public class EmailProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public void sendEmail(String message) {
        // "daily-email-job"이라는 토픽에 메시지 전송
        kafkaTemplate.send("daily-email-job", message);
        System.out.println("kafka로 메시지 전송 완료: " + message);
    }
}

3.5. Consumer 구현 (받는 쪽)​

@KafkaListener 어노테이션을 쓴다.

@Service
public class EmailConsumer {

    @KafkaListener(topics = "daily-email-job", groupId = "email-consumer-group")
    public void consume(String message) {
        System.out.println("Kafka에서 메시지 수신: " + message);

        // 실제 이메일 발송 로직 수행 (여기서 외부 API 호출)
        sendToExternalApi(message);
    }
}

4. 트러블 슈팅 : 메시지를 안 읽어오는 현상​

프로젝트를 진행하며 겪은 문제는 Consumer가 데이터를 읽어오지 않는 현상이었다.

원인은 auto.offset.reset 기본값이 latest라서 그렇다. Producer가 1만 개를 다 보낸 뒤 Consumer를 켜면, Consumer는 "내가 오기 전 일은 몰라"하고 아무것도 읽지 않는다.

earliest로 바꾸거나 Consumer를 먼저 켜두는 것으로 해결했다.