Kafka 개념과 Spring Boot + Kafka 간단한 연동

서론

기존 데이터 시스템의 구조는 각 애플리케이션과 데이터베이스가 end-to-end로 직접 연결되어 있었습니다. 이러한 구조는 간단하지만 각각의 데이터 파이프라인이 분리되어 있어, 요구사항이 증가함에 따라 시스템의 복잡도를 높이는 결과를 가져왔고, 크게 아래와 같은 문제점들이 발생했습니다.

시스템 복잡도의 증가

• 중앙화된 데이터 전송 영역이 없어, 데이터의 흐름을 파악하기 어렵고, 시스템 관리가 복잡함.
• 시스템의 일부분에 문제가 발생하면, 연결된 모든 애플리케이션들을 확인해야 함.

데이터 일관성 유지의 어려움

• 데이터가 여러 시스템과 데이터베이스에 분산되어 있는 경우, 한 시스템에서 변경된 데이터가 다른 시스템에 즉시 반영되지 않아 데이터의 일관성을 유지하기 어려움.

데이터 실시간 처리의 어려움

• 전통적인 메시지 큐 시스템이나 데이터베이스는 대부분 배치 처리 방식을 사용함.
• 이는 데이터를 실시간으로 처리하는 것이 어렵다는 것을 의미.

확장성 제한

• 대부분의 전통적인 메시지 큐 시스템은 한정된 리소스 내에서 작동하므로, 대량의 데이터를 처리하는 데 제한이 있음.
• 데이터의 양이 증가하면서 시스템을 확장해야 하는 상황에서 이런 제한이 큰 문제가 될 수 있음.

아파치 카프카(Apache Kafka)는 이런 문제점들을 해결하기 위해 링크드인에서 개발되었고, 현재는 Apache Software Foundation의 오픈 소스 프로젝트로 유지 관리되는 분산 스트리밍 플랫폼입니다. 

 

 

Kafka의 기본 구조

Kafka는 크게 아래의 네 가지의 핵심 구성 요소로 이루어져 있습니다.

1. 프로듀서(Producer)

프로듀서는 Kafka에 메시지를 발행하는 역할을 하는 컴포넌트입니다.

프로듀서는 다양한 데이터 소스로부터 데이터를 가져와 Kafka의 특정 토픽에 메시지를 발행합니다.

2. 브로커(Broker)

브로커는 Kafka의 핵심 서버 컴포넌트로, 프로듀서로부터 메시지를 받아서 저장하고, 컨슈머에게 메시지를 전달하는 역할을 합니다.

Kafka 클러스터는 여러 브로커들로 구성되며, 각 브로커는 하나 이상의 토픽의 메시지를 저장하고 관리합니다.

3. 토픽(Topic)

토픽은 Kafka에서 데이터를 분류하는 단위입니다.

프로듀서는 메시지를 특정 토픽에 발행하고, 컨슈머는 토픽을 구독하여 메시지를 소비합니다. 토픽은 여러 파티션으로 나뉘어질 수 있고, 이를 통해 데이터를 병렬로 처리할 수 있습니다. 각 파티션은 순서가 보장된 메시지 스트림을 제공하며, 브로커가 클러스터 내에서 파티션을 분산하여 저장합니다.

4. 컨슈머(Consumer)

컨슈머는 Kafka의 특정 토픽을 구독하고, 해당 토픽의 메시지를 소비하는 역할을 하는 컴포넌트입니다.

컨슈머는 하나 이상의 토픽을 구독할 수 있으며, 토픽의 파티션을 동시에 소비할 수 있습니다.

 

 

Kafka와 Zookeeper의 관계

https://ifuwanna.tistory.com/487

Kafka는 Apache Zookeeper를 사용하여 클러스터를 관리합니다. Zookeeper는 브로커, 토픽, 파티션 등 Kafka의 메타데이터 정보를 저장하고 관리하여 Kafka 클러스터의 상태를 일관되게 유지하고, 브로커의 실패 및 복구를 감지하는 등의 역할을 수행합니다. 따라서 Kafka를 띄우기 위해서는 Zookeeper가 반드시 실행되어야 합니다.

하지만 파티션이 100,000개가 넘었을 때 Kafka 클러스터에 스케일링 문제가 발생했고, 유지보수와 설정 그리고 안정성 개선을 위해 최근 Kafka는 Zookeeper의 의존성을 줄이기 위한 노력을 하고 있습니다.

실제로 Kafka 3.x부터는 Zookeeper 없이 카프카를 구동할 수 있는 Kraft(Kafka Raft Metadata mode) 모드가 추가되었으며 Kafka 4.0부터는 Zookeeper의 의존성을 완전히 지운 오직 KRaft만 지원되는 버전으로 출시될 예정입니다.(https://www.confluent.io/blog/kafka-without-zookeeper-a-sneak-peek/)

 

 

RabbitMQ와의 차이

RabbitMQ와 Kafka의 가장 큰 차이점은 RabbitMQ는 전통적인 메시지 브로커(Message Broker)이고, Kafka는 이벤트 스트리밍 플랫폼(Event Streaming Platform)이라는 점입니다.

위와 같이 발행자(Publisher)가 Message Exchange에 메시지를 보내면, 구독자(Subscriber)가 메시지를 구독합니다. 여러 개의 Publisher가 Message Exchange에 메시지를 보내면, 정해진 규칙에 따라서 큐에 Routing이 되고, Consumer들이 메시지를 처리합니다.(여기서 Kafka와 다른 점은 Kafka는 큐를 구현하지 않는다는 점입니다. Kafka는 위에서 봤듯이 토픽(Topic)으로 불리는 카테고리에 데이터 집합을 저장하며, 각각의 토픽에 분할된 메시지 로그를 가지고 있습니다.)

그리고 컨슈머가 메시지를 가져가면 큐에는 더 이상 남지 않고 사라지게 됩니다. 이러한 전통적인 메시지 브로커의 형태는 소비자와 메시지 브로커의 결합력이 높아지게 되어 후에 서비스의 트래픽이 증가하였을 때 수평적으로 확장하는 데에 어려움이 있었으며, 이벤트 메시지가 성공적으로 전달되었다고 판단될 경우 이 메시지는 큐에서 삭제되어 버리기 때문에 후에 다시 이벤트를 받기가 어렵다는 단점이 존재했습니다.

카프카는 이러한 단점을 극복하며 생긴 이벤트 스트리밍 플랫폼으로써, 메시지 브로커와 이벤트 스트리밍 플랫폼 모두 이벤트를 수신하고, 이것을 소비자에게 전달하는 데에 목적을 두고 있지만 작동 방식에 큰 차이가 있습니다.

이벤트 스트리밍 플랫폼은 메시지 브로커와 다르게 토픽(Topic)이라는 것이 Event Streamer에 저장됩니다. Event producer가 이벤트를 생성하면, 토픽이라고 불리는 이벤트의 레코드 로그를 streamer에 순서대로 기록하게 됩니다. 그 후 해당 토픽을 구독(subscribe)한 컨슈머에게 전달하게 됩니다. 또한 이 토픽을 컨슈머가 가져간 후에도 이벤트 스트림에서 계속 토픽을 유지하기 때문에 장애 발생 상황에서도 이벤트를 다시 발행할 수 있습니다.

Kafka vs RabbitMQ	RabbitMQ	Kafka
성능	초당 4,000 ~ 10,000개의 메시지 처리 가능	초당 1,000,000 개의 메시지 처리 가능
메시지 생명주기	소비되는 순간 삭제	정책 기반(설정 한 기간만큼)
트랜잭션 데이터	중복 메시지 발송 및 메시지 손실 가능성 존재	순서 보장 및 메시지 유실 가능성이 적음
운영 데이터	라우팅, 클러스터링, 메시지 확인 등의 운영 데이터 처리에 강점	대량의 실시간 로그 또는 스트림 데이터 처리에 강점
클러스터링	클러스터링 지원, 노드 간 메시지 복제로 고가용성 제공	클러스터링 지원, 자동 복제와 파티셔닝을 통해 고가용성 및 장애 복구 기능 제공
메시지 모델	Queue와 Topic 모델 지원	Topic 모델만 지원
스케일링	스케일 업에 적합	스케일 아웃에 적합, 대량의 데이터를 처리할 수 있는 높은 확장성

더 상세한 차이점은 링크에서 확인 가능합니다.

 

 

Spring Boot + Kafka 연동

테스트 환경은 다음과 같습니다.

• Spring Boot 2.7 + Java 11 + gradle
• Docker kafka & Zookeeper

깃허브: https://github.com/dkswnkk/Spring-Playground/tree/main/kafka-run-practice

Producer Application Project

1. 의존성 추가

// kafka
implementation 'org.springframework.kafka:spring-kafka'

 

2. Producer Config 설정

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

설정된 속성에 대한 설명은 다음과 같습니다.

• BOOTSTRAP_SERVERS_CONFIG: Producer가 처음으로 연결할 Kafka 브로커의 위치를 설정합니다. 현재 코드의 경우 localhost의 9092 포트에 위치하도록 설정했습니다.
• KEY_SERIALIZER_CLASS_CONFIG & VALUE_SERIALIZER_CLASS_CONFIG: Producer가 Key와 Value 값의 데이터를 Kafka 브로커로 전송하기 전에 데이터를 byte array로 변환하는 데 사용하는 직렬화 메커니즘을 설정합니다. Kafka는 네트워크를 통해 데이터를 전송하기 때문에, 객체를 byte array로 변환하는 직렬화 과정이 필요합니다. 따라서, StringSerializer를 사용해 직렬화했습니다.

KafkaTemplate는 Spring Kafka에서 제공하는 Kafka Producer를 Wrapping 한 클래스입니다. KafkaTemplate는 Kafka에 메시지를 보내는 여러 메서드를 제공하며, 이 메서드를 사용하여 브로커로 메시지를 보내기 위해 직접 Kafka Producer API를 사용하는 대신, send와 같은 메서드를 통해 더 편리하고 간결한 코드로 메시지를 보낼 수 있습니다.

3. Producer 생성

package com.example.kafkarunpractice.producer;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class TestProducer {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public TestProducer(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void create() {
        kafkaTemplate.send("topic", "say hello~");
    }
}

 

 

Consumer Application Project

1. 의존성 추가

// kafka
implementation 'org.springframework.kafka:spring-kafka'

2. Consumer Config 설정

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_1");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        return factory;
    }
}

ConsumerConfig.GROUP_ID_CONFIG는 Consumer가 속한 Consumer 그룹의 ID를 설정합니다. Consumer 그룹은 같은 토픽을 소비하는 Consumer들의 그룹으로, 그룹 내의 모든 Consumer는 토픽의 서로 다른 파티션에서 메시지를 읽어 들입니다.

이를 통해 메시지 처리를 병렬화 하여 처리 속도를 향상시킬 수 있으며, Consumer가 실패할 경우 다른 Consumer가 해당 Consumer의 파티션을 처리하여 고가용성을 제공할 수 있습니다. ConcurrentKafkaListenerContainerFactory는 Spring의 @KafkaListener 어노테이션이 붙은 메서드에 주입되어 사용되며, 메시지를 동시에 처리할 수 있는 메시지 리스너 컨테이너를 생성합니다.

3. Consumer 생성

@Component
public class TestConsumer {

    @KafkaListener(topics = "topic", groupId = "group_1")
    public void listener(Object data) {
        System.out.println(data);
    }
}

 

 

Kafka & Zookeeper 실행

docker-compose.yml 설정

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:2.12-2.5.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

Docker Compose를 이용하여 Kafka와 Zookeeper를 도커 컨테이너로 실행시킵니다. Zookeeper는 기본적으로 2181 포트를 사용합니다.

• 컴포즈 실행: docker-compose up -d
• 컴포즈 종료: docker-compose down

 

테스트

Consumer Application Project를 실행하고, Produce Application Project에서 생성한 아래의 테스트 메서드를 실행했습니다.

@SpringBootTest
class TestProducerTest {

    @Autowired
    private TestProducer testProducer;

    @Test
    void test() {
        testProducer.create();
    }
}

수행결과

정상적으로 Producer Application에서 발행한 토픽이 Consumer Application에서 소비되는 걸 확인할 수 있습니다.

 

 

같이 보면 좋은 글

• [Kafka] 리밸런싱 종류와 파티션 할당 전략
• [Kafka] 프로듀서 멱등성 보장하기
• [Kafka] 컨슈머 오프셋 수동으로 커밋하기
• Spring Boot에서 여러 Kafka 클러스터 사용하기