Kafka란?

🔍1. 이벤트 기반 아키텍처(Event-Driven Architecture)란?

분산된 애플리케이션 서비스들이 이벤트를 기반으로 통신하고 서로의 동작을 야기하는 패턴을 이벤트 기반 아키텍처라고 한다.

• 서비스들이 이벤트를 발행(publish)하고, 다른 서비스가 이를 구독(subscribe)하여 처리.
• Kafka, RabbitMQ, SQS, EventBridge 등과 같은 메시징 시스템을 통해 구현.

⭐ 이벤트 기반 아키텍처를 사용하는 이유

1️⃣ 서비스 간 결합도↓ → 각 서비스가 독립적으로 동작 가능 (확장성↑)
2️⃣ 비동기 처리 → 빠른 응답 가능, 대량 트래픽 처리에 유리
3️⃣ 확장성 & 유연성 → 새로운 기능 추가 시 기존 시스템에 영향 최소화 

 

📌 2. Kafka란?

Kafka는 대규모 데이터를 실시간으로 처리하고 저장하는 분산형 메시징 시스템이자 스트리밍 플랫폼이다. 주로 이벤트 기반 아키텍처, 실시간 데이터 파이프라인, 로그 처리 등에 활용되며, MSA 환경에서 비동기 데이터 전달 및 처리를 위한 강력한 도구로 사용된다.

✅ 2-1. 아키텍처

Kafka는 분산형 메시지 큐 또는 실시간 스트리밍 플랫폼으로 동작하며, 데이터가 다음과 같은 경로로 전송된다.

Producer → Kafka 브로커 → 파티션 → Consumer

 

✅ 2-2. Kafka의 주요 특징

• 분산 시스템: 여러 개의 브로커로 구성되어 확장성, 내결함성, 안정성 제공
• 실시간 데이터 처리: 이벤트 기반 아키텍처 및 실시간 분석 시스템에 최적화
• 내구성: 데이터를 디스크에 저장하고 복제(Replication)하여 장애 발생 시에도 데이터 보호
• 메시지 큐 역할: Producer → Topic → Consumer 구조로 동작하며, 비동기 방식 지원
• 확장성: 데이터를 *파티셔닝(Partitioning) 하여 수평 확장 가능

 

⭐샤딩 vs 파티셔닝

파티셔닝은 하나의 서버, 샤딩은 여러 서버에 분산한다는 차이

개념	설명	Kafka에서의 역할
파티셔닝	하나의 토픽을 여러 개의 논리적 단위(파티션)로 나눔	토픽 데이터를 분산 처리, Consumer 병렬 처리 가능
샤딩	데이터를 여러 서버(노드)로 분산 저장	Kafka에서는 브로커가 샤딩 개념과 유사, 여러 브로커가 데이터 분산 처리

 

📌 3. Kafka의 기본 개념 & 명령어

✅ 3-1. Kafka 브로커

• Kafka 클러스터 내의 서버를 지칭, 데이터 저장 및 관리, 메시지 전달 역할을 함.
• 여러 브로커가 함께 작동하여 고가용성과 확장성을 제공.

 

✅ 3-2. 토픽 (Topic) - ‘메시지’라고 자주 표현

• 데이터의 분류 단위로, Producer는 특정 토픽에 데이터를 전송하고, Consumer는 해당 토픽에서 데이터를 수신.
• 파티션(Partition) 단위로 나뉘며, 각 파티션은 메시지 순서 보장

 

✔️ 토픽 관리 명령어

# 토픽 생성
bin/kafka-topics.sh --create --topic <토픽명> --bootstrap-server <브로커주소> --partitions 3 --replication-factor 1

# 토픽 리스트 조회
bin/kafka-topics.sh --list --bootstrap-server <브로커주소>

# 토픽 상세 정보 조회
bin/kafka-topics.sh --describe --topic <토픽명> --bootstrap-server <브로커주소>

# 토픽 삭제
bin/kafka-topics.sh --delete --topic <토픽명> --bootstrap-server <브로커주소>

 

⚠️ 토픽 삭제 활성화 필요
Kafka 설정 파일 (server.properties)에서 아래 옵션 설정 필요

delete.topic.enable=true //이거 설정해야 삭제 가능
auto.create.topics.enable=true //Provider에서 토픽 자동 생성

 

✅ 3-3. 이벤트 (Event)

• Kafka에서 이벤트는 데이터의 상태 변화나 동작을 나타내는 메시지
• 기본적으로 개별 이벤트 삭제 불가 ❌

 

✔️ Retention Policy 설정 (이벤트 자동 삭제)

• 토픽 단위로 메시지 보존 기간(retention.ms) 또는 최대 크기(retention.bytes)를 설정해 오래된 메시지를 자동 삭제

log.retention.ms=604800000  # 7일 동안 보존
log.retention.bytes=1073741824  # 1GB 초과 시 삭제

 

✅ 3-4. 파티션 (Partition)

• 토픽의 세분화 단위. 1개의 토픽은 1개 이상의 파티션으로 구성, 분산 저장되어 병렬 처리가 가능.
• 파티션 수는 병렬 처리 성능에 영향을 미치므로, 클라이언트와 브로커의 처리량을 고려해 결정.

 

⭐ 리플리케이션 (Replication)

• 데이터 복제를 통해 장애 발생 시 데이터 보호.
• 파티션의 복제본 (하나의 리더 파티션과 여러 팔로워 파티션으로 구성.), 브로커 개수보다 1개 적게 설정하는게 일반적

 

✔ 토픽 생성 시 파티션 & 리플리케이션 설정

bin/kafka-topics.sh --create --topic my-topic --partitions <파티션 수> --replication-factor 2 --bootstrap-server localhost:9092

 

✅ 3-5. 오프셋 (Offset)

• 각 파티션에서 메시지의 고유 ID(번호)
• Consumer는 오프셋을 기준으로 메시지 순서대로 처리
• Consumer Group별로 오프셋 관리 가능(메시지 처리 상태 추적 등)

 

✔️ 오프셋 리셋 (이전 메시지 다시 읽고 싶을 때)

bin/kafka-consumer-groups.sh --bootstrap-server <브로커주소> --group <그룹명> --reset-offsets --topic <토픽명> --to-earliest --execute

 

✔️ 오프셋 개별 삭제

• Kafka는 오프셋을 개별적으로 삭제할 수 있는 기능을 제공하지 않습니다.
• 특정 오프셋을 삭제하는 대신 전체 토픽을 삭제하거나, Consumer Group 자체를 재구성하여 새로운 오프셋을 생성할 수 있습니다.

 

✅ 3-6. Producer (데이터 전송)

• Kafka에 데이터를 전송하는 클라이언트.
• 메시지를 특정 토픽의 특정 파티션에 전송.

 

✔️ 이벤트 발행 (Producer 역할)

bin/kafka-console-producer.sh --topic <토픽명> --bootstrap-server <브로커주소>

 

✅ 3-7. Consumer (데이터 소비)

• Kafka에서 데이터를 소비하는 클라이언트.

 

✔️ 이벤트 확인 (Consumer 역할)

bin/kafka-console-consumer.sh --topic <토픽명> --bootstrap-server <브로커주소> --from-beginning

 

✅ 3-8. Consumer 그룹 (Consumer Group)

• 여러 Consumer로 구성된 논리적 단위
• Consumer 그룹 당 각 파티션을 한 번만 읽음, 같은 그룹의 Consumer들은 메시지를 분산 처리
• 서로 다른 Consumer 그룹은 동일 데이터를 독립적으로 소비 가능

 

✔️ Consumer Group 관리 명령어

# Consumer Group 리스트 확인
bin/kafka-consumer-groups.sh --bootstrap-server <브로커주소> --list

# Consumer Group 상세 정보 조회
bin/kafka-consumer-groups.sh --bootstrap-server <브로커주소> --describe --group <그룹명>

# Consumer Group 삭제 (오프셋 포함)
bin/kafka-consumer-groups.sh --bootstrap-server <브로커주소> --group <기존 그룹 이름> --delete

 

📌 4. Kafka 설치 및 실행

Kafka는 메시지 브로커이다. 프로젝트에서 Kafka를 사용하려면

1. Kafka 브로커가 메시지를 관리.
2. Zookeeper가 Kafka 브로커를 조율.
3. 애플리케이션(Spring Kafka)가 Kafka 브로커와 통신

 

⭐ ZooKeeper: 브로커를 관리하고, 리더 선출과 메타데이터를 저장. (Kafka 2.8 이후에는 Kafka 자체 메타데이터 관리로 대체 가능)

 

✅ 4-1. Kafka 서버 (브로커)

✅ 4-1-1. Kafka 서버 설치

✔️  Kafka 다운로드

Apache Kafka 다운로드 링크

 

✔️  소켓 서버 리스너 설정

server.properties 파일을 열어서 “listeners=PLAINTEXT://:9092” 앞에 붙은 주석(#)을 제거해 주면 됩니다.

 

✅ 4-1-2. Kafka 실행

Kafka는 Zookeeper와 함께 실행해야 한다. 다운로드한 Kafka 디렉터리에서 실행.

*Bat파일이 없는데 window에서는 git bash에서 실행하면 됩니다!!

 

1️⃣ Zookeeper 실행

Documents/Tools/kafka_2.13-3.9.0에서

bin/zookeeper-server-start.sh config/zookeeper.properties

 

2️⃣  Kafka 브로커 실행

bin/kafka-server-start.sh config/server.properties

 

3️⃣ Kafka 브로커 확인

Kafka 브로커가 실행되면 기본적으로 localhost:9092에서 요청을 대기 중입니다.

 

✅ 4-2. Kafka 클라이언트

1️⃣ 의존성 추가

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

 

2️⃣ Application.properties

처음에는 필수인 것 만 해보자!!

 

📖  Kafka Consumer 설정

# Kafka 브로커 리스트 (필수)
# 브로커 호스트 및 포트 정보. 여러 브로커는 쉼표로 구분.
spring.kafka.bootstrap-servers=localhost:9092

# Consumer 그룹 ID 설정 (필수)
# 동일한 그룹 ID를 가진 Consumer는 메시지를 공유(로드 밸런싱)받음.
spring.kafka.consumer.group-id=order-service-group

# Consumer Key 역직렬화 클래스 설정 (필수)
# 메시지 키를 디코딩할 때 사용. 기본적으로 StringDeserializer 사용.
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer

# Consumer Value 역직렬화 클래스 설정 (필수)
# 메시지 값을 디코딩할 때 사용. 기본적으로 StringDeserializer 사용.
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

# 메시지 자동 커밋 여부 (Optional)
# true: Consumer가 메시지 처리 후 자동으로 오프셋을 커밋.
spring.kafka.consumer.enable-auto-commit=true

# 자동 커밋 주기 (Optional)
# 메시지 처리 완료 후 오프셋을 자동으로 커밋하는 간격 (밀리초 단위).
spring.kafka.consumer.auto-commit-interval=100

# Consumer가 처음 시작할 때 읽기 시작할 오프셋 설정 (Optional)
# latest: 가장 최근의 메시지부터 읽기 시작.
# earliest: 토픽의 처음 메시지부터 읽기 시작.
spring.kafka.consumer.auto-offset-reset=latest

# Consumer 세션 타임아웃 (Optional)
# Consumer가 브로커와의 연결이 끊겼다고 판단하기까지의 시간 (밀리초 단위).
spring.kafka.consumer.properties.session.timeout.ms=15000

# 하트비트 간격 (Optional)
# Consumer가 브로커와 연결을 유지하기 위해 하트비트를 보내는 간격 (밀리초 단위).
spring.kafka.consumer.properties.heartbeat.interval.ms=5000

 

📖 Kafka Producer 설정

# Producer Key 직렬화 클래스 설정 (필수)
# 메시지 키를 인코딩할 때 사용. 기본적으로 StringSerializer 사용.
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer

# Producer Value 직렬화 클래스 설정 (필수)
# 메시지 값을 인코딩할 때 사용. 기본적으로 StringSerializer 사용.
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

# 메시지 전송 확인 모드 (Optional)
# 0: 브로커 확인 없음.
# 1: 리더 파티션에 기록 성공 확인.
# all: 모든 파티션 복제본에 기록 성공 확인.
spring.kafka.producer.acks=1

# 전송 실패 시 재시도 횟수 (Optional)
# 메시지 전송이 실패했을 때 재시도 횟수 설정.
spring.kafka.producer.retries=3

# 메시지 배치 크기 (Optional)
# 프로듀서가 한 번에 전송할 메시지 배치 크기 (바이트 단위).
spring.kafka.producer.batch-size=16384

# 배치 전송 대기 시간 (Optional)
# 배치가 꽉 차지 않아도 일정 시간 대기 후 전송 (밀리초 단위).
spring.kafka.producer.linger-ms=1

# Producer 메모리 버퍼 크기 (Optional)
# 전송 전 메시지를 저장하는 메모리 버퍼 크기 (바이트 단위).
spring.kafka.producer.buffer-memory=33554432

# --------------------------------------------
# 자주 사용하는 추가 설정
# --------------------------------------------

# 메시지 최대 크기 (Optional)
# Producer와 Consumer에서 전송/수신 가능한 메시지의 최대 크기 (바이트 단위).
spring.kafka.producer.properties.max.request.size=1048576
spring.kafka.consumer.properties.fetch.max.bytes=1048576

# 토픽별 오프셋 커밋 주기 (Optional)
# 자동 커밋을 사용할 경우 오프셋을 커밋하는 주기 (밀리초 단위).
spring.kafka.consumer.properties.auto.commit.interval.ms=5000

# SSL 설정 (Optional)
# SSL 연결이 필요한 경우 설정.
spring.kafka.properties.ssl.truststore.location=/path/to/truststore
spring.kafka.properties.ssl.truststore.password=yourpassword
spring.kafka.properties.ssl.keystore.location=/path/to/keystore
spring.kafka.properties.ssl.keystore.password=yourpassword
spring.kafka.properties.security.protocol=SSL

# 요청 시간 초과 (Optional)
# 브로커가 요청을 처리하지 못하면 초과 시 실패 처리 (밀리초 단위).
spring.kafka.producer.properties.request.timeout.ms=30000
spring.kafka.consumer.properties.request.timeout.ms=30000

# 로그 수준 설정 (Optional)
# Kafka 클라이언트 로그 수준 설정.
logging.level.org.apache.kafka=INFO

 

📌 5. Kafka 이벤트 사용 예시

✅ 5-1. 이벤트 객체 생성

package com.example.order_service.event;
import lombok.Data;

@Data
public class OrderCreatedEvent {
    private int orderId;
    private int productId;
    private int colorId;
    private int sizeId;
    private int quantity;
    private int userId;
}

 

✅ 5-2. Event 발급

📖  kafkaTemplate.send("order-created-topic", event);

// OrderService.java

public Orders createOrder(OrderRequest request) {
        // 주문 객체 생성
        Orders order = dtoToOrder(request, OrderStatus.CREATED);

        // 주문 상품 리스트
        List<OrderItems> orderItems = new ArrayList<>();
        for (OrderItemRequest itemRequest : request.getItems()) {
            OrderItems orderItem = dtoToOrderItems(itemRequest, order);
            orderItems.add(orderItem);
        }
        order.setOrderItems(orderItems);

        // 총 금액 계산
        BigDecimal totalPrice = orderItems.stream()
                .map(item -> item.getPrice().multiply(new BigDecimal(item.getQuantity())))
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        order.setTotalPrice(totalPrice);

        // 주문 저장
        Orders savedOrder = orderRepository.save(order);

        // 이벤트 발행
        OrderCreatedEvent event = orderToEvent(order);
        kafkaTemplate.send("order-created-topic", event);
        return savedOrder;
}

 

* 요청 예시

Post http://localhost:8080/order/api/orders

{
    "userId": 3,
    "shippingAddress": "서울시",
    "paymentMethod": "card",
    "items": [{
        "productId": 1,
        "sizeId": 1,
        "colorId": 1,
        "quantity": 1,
        "price": 199.99
    }]
}

 

✅ 5-3. Event 수신

 

📖  @KafkaListener(topics = "order-created-topic", groupId = "products-service-group")

//OrderEventListener.java

@KafkaListener(topics = "order-created-topic", groupId = "products-service-group")
public void handleOrderCreatedEvent(String message) {
    try {
        // JSON 메시지를 OrderEvent 객체로 변환
        OrderEvent orderEvent = objectMapper.readValue(message, OrderEvent.class);

        // 색상과 사이즈에 따른 재고 조회
        Optional<ProductStocks> optionalStock  = productStocksRepository.findStockQuantity(
            orderEvent.getProductId(),
            orderEvent.getColorId(),
            orderEvent.getSizeId()
        );

        if (optionalStock.isPresent()) {
            ProductStocks stock = optionalStock.get();
            int updatedStock = stock.getStockQuantity() - orderEvent.getQuantity();

            // 재고 부족 시 예외 처리
            if (updatedStock < 0) {
                throw new IllegalStateException("재고 부족: 요청한 수량을 처리할 수 없습니다.");
            }   

            // 재고 수량 업데이트
            stock.setStockQuantity(updatedStock);
            productStocksRepository.save(stock);   

            System.out.println("재고 업데이트 성공");                
        } else {
            System.out.println("해당 색상/사이즈에 대한 재고 정보가 없습니다.");
        }
    } catch (Exception e) {
        System.out.println("주문 처리 중 오류 발생: " + e.getMessage());
    }
}

 

⭐  6. 알아두면 좋은 설정 & 명령어

1. 주요 스크립트

Kafka의 명령어는 Kafka 설치 디렉토리의 bin/ 폴더에 있는 스크립트를 사용.

• kafka-topics.sh: 토픽 생성, 조회, 삭제 등을 관리.
• kafka-console-producer.sh: 메시지 발행(Producer 역할).
• kafka-console-consumer.sh: 메시지 확인(Consumer 역할).
• kafka-configs.sh: 브로커나 토픽 설정 변경.

 

2. 메시지 확인용 ACL 비활성화 (개발/테스트용)

Kafka는 보안 설정이 기본 활성화된 경우 Consumer/Producer가 인증되지 않으면 실패할 수 있습니다. 개발용으로 sasl.mechanism을 설정하지 않으려면 server.properties에서 아래 옵션을 확인하세요:

authorizer.class.name=

 

3. 브로커 상태 확인

bin/kafka-broker-api-versions.sh --bootstrap-server <브로커주소>

 

4. 메시지 확인 시 JSON 포맷 파싱

bin/kafka-console-consumer.sh --topic <토픽명> --bootstrap-server <브로커주소> --from-beginning | jq

 

5. 개발 환경에서는 Kafka 비활성화

kafka.enabled=false

spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration

 

@ConditionalOnProperty(name = "kafka.enabled", havingValue = "true", matchIfMissing = true)