Kafka를 활용한 CQRS (1) - Kafka 기초와 설치

게시 2026/08/13

By eta_kyung

22 분읽는 시간

1. 개요

여러 애플리케이션이 서로 직접 데이터를 주고받는 구조에서는 파이프라인 개수가 늘어날수록 소스 코드와 버전 관리가 복잡해지고, 타깃 애플리케이션 하나에 장애가 나면 그 영향이 소스 애플리케이션까지 그대로 전파되는 문제가 생긴다. Kafka는 이런 문제를 해결하기 위해 LinkedIn 내부 데이터 흐름을 개선할 목적으로 개발된 오픈소스 메시지 브로커로, 애플리케이션끼리 직접 연결하는 대신 한 곳에 데이터를 모아 처리하는 중앙 집중형 구조를 가능하게 해준다. 이번 글에서는 Kafka의 기본 개념과 설치 방법, 파이썬 클라이언트 사용법까지 정리한다.

Kafka 내부에서 데이터가 쌓이는 파티션은 큐 자료구조처럼 FIFO(First In First Out) 방식으로 동작한다. 이 큐에 데이터를 밀어 넣는 쪽이 프로듀서, 큐에서 데이터를 꺼내가는 쪽이 컨슈머다. Kafka는 자바로 만들어져 자바에서 다룰 수 있는 대부분의 객체를 다룰 수 있으며, 상용 환경에서는 최소 3대 이상의 서버(브로커)로 분산 운영해 프로듀서가 보낸 데이터를 파일 시스템에 안전하게 기록한다. 빅데이터나 데이터 파이프라인을 구축하는 곳에서는 사실상 필수 요소로 자리 잡았다.

2. Kafka를 사용하는 이유

특징	설명
높은 처리량	대량의 데이터를 지연 없이 빠르게 처리할 수 있다
확장성	브로커를 늘리거나 줄이는 스케일 아웃/인이 비교적 쉽다
영속성	메모리가 아니라 파일 시스템에 데이터를 저장해 유실 위험이 낮다
고가용성	여러 브로커에 데이터를 복제해 일부 장애에도 서비스가 유지된다

3. 데이터 레이크 아키텍처

데이터를 모으고 처리하는 방식은 시대에 따라 다음과 같이 발전해왔다.

전통적인 방식: 서비스 애플리케이션 각각이 End-to-End로 데이터를 직접 모아 처리하는 구조
람다 아키텍처: 배치 처리와 실시간 처리를 별도 경로로 병행하는 구조
카파 아키텍처: 배치 경로 없이 스트리밍 처리 하나로 통합한 구조
스트리밍 데이터 레이크 아키텍처: Kafka 같은 스트리밍 플랫폼을 중심에 두고 데이터를 실시간으로 적재·처리하는 구조

Kafka는 이 중 스트리밍 기반 아키텍처를 구현하는 핵심 구성 요소로 쓰인다.

4. 설치

4.1 Docker로 설치

docker-compose.yml 파일을 만들어 주키퍼와 Kafka를 함께 띄운다.

  
version: '3.8'
services:
  zookeeper:
    image: wurstmeister/zookeeper:latest
    container_name: zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:latest
    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 up -d

토픽 생성/조회/삭제

컨테이너 내부에 접속한 뒤 Kafka 명령어가 있는 디렉토리로 이동해서 작업한다.

  
# 컨테이너 접속
docker exec -it kafka /bin/bash

# 명령어 디렉토리로 이동
cd /opt/kafka/bin

# 토픽 생성
kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic exam-topic

# 토픽 목록 확인
kafka-topics.sh --bootstrap-server localhost:9092 --list

# 토픽 삭제
kafka-topics.sh --delete --zookeeper zookeeper:2181 --topic exam-topic

토픽 생성 명령어의 옵션을 하나씩 뜯어보면 다음과 같다.

옵션	의미
`kafka-topics.sh`	토픽을 생성/조회/삭제할 때 쓰는 Kafka 제공 스크립트
`--create`	토픽을 새로 만들겠다는 동작 지정
`--zookeeper zookeeper:2181`	어떤 Zookeeper에 접속해서 이 작업을 처리할지 지정 (`zookeeper`는 호스트명, `2181`은 Zookeeper 기본 포트)
`--replication-factor 1`	이 토픽의 데이터를 몇 개의 브로커에 복제해둘지 지정. `1`이면 복제본 없이 원본 하나만 존재해서 장애에 취약하지만 테스트/로컬 환경에서는 충분하다
`--partitions 1`	이 토픽을 몇 개의 파티션으로 나눌지 지정. `1`이면 병렬 처리 없이 파티션 하나로만 동작한다
`--topic exam-topic`	생성할 토픽의 이름

참고로 --zookeeper 옵션은 구버전 방식이고, 최신 Kafka에서는 브로커에 직접 접속하는 --bootstrap-server 방식(토픽 조회 명령어처럼)을 더 많이 쓴다. 실무에서는 보통 하나로 통일해서 --bootstrap-server 방식을 사용하는 편이다.

토픽 목록 확인 명령어(kafka-topics.sh --bootstrap-server localhost:9092 --list)의 옵션도 정리하면 다음과 같다.

옵션	의미
`kafka-topics.sh`	토픽 관련 작업(생성/조회/삭제)을 수행하는 스크립트
`--bootstrap-server localhost:9092`	접속할 Kafka 브로커의 주소. Zookeeper가 아니라 브로커에 직접 접속한다
`--list`	존재하는 토픽 이름을 전부 나열하라는 동작 지정

같은 kafka-topics.sh 스크립트인데 생성 명령어는 --zookeeper, 조회 명령어는 --bootstrap-server를 쓰는 이유는 접속 대상이 다르기 때문이다. 예전 Kafka는 메타데이터 관리를 Zookeeper가 전담해서 --zookeeper 옵션만 지원했지만, 이후 버전부터 브로커가 이런 요청을 직접 처리할 수 있게 되면서 --bootstrap-server 옵션이 추가되었다. 지금은 두 방식이 혼용되고 있지만, 최신 버전에서는 --bootstrap-server로 통일해서 쓰는 것이 권장된다.

메시지 전송/수신 확인

  
# 프로듀서 콘솔 실행
kafka-console-producer.sh --topic exam-topic --broker-list localhost:9092

# (다른 터미널에서 컨테이너 접속 후) 컨슈머 콘솔 실행
kafka-console-consumer.sh --topic exam-topic --bootstrap-server localhost:9092 --from-beginning

프로듀서/컨슈머 콘솔 명령어의 옵션도 정리하면 다음과 같다.

프로듀서 콘솔

옵션	의미
`kafka-console-producer.sh`	터미널에서 직접 메시지를 입력해 토픽으로 보내주는 콘솔용 프로듀서 스크립트
`--topic exam-topic`	메시지를 보낼 대상 토픽 이름
`--broker-list localhost:9092`	접속할 브로커 주소 목록. 프로듀서는 이 브로커에 직접 연결해서 메시지를 전달한다

실행하면 입력을 기다리는 프롬프트가 뜨고, 한 줄씩 입력하고 엔터를 칠 때마다 그 줄이 하나의 메시지로 exam-topic에 전송된다.

컨슈머 콘솔

옵션	의미
`kafka-console-consumer.sh`	토픽에 쌓인 메시지를 터미널에 실시간으로 출력해주는 콘솔용 컨슈머 스크립트
`--topic exam-topic`	메시지를 읽어올 대상 토픽 이름
`--bootstrap-server localhost:9092`	접속할 브로커 주소. 컨슈머도 브로커에 직접 연결해서 메시지를 가져온다
`--from-beginning`	컨슈머를 실행한 시점 이후의 새 메시지만 보는 게 아니라, 토픽에 남아있는 가장 오래된 메시지부터 전부 읽어오라는 옵션

프로듀서 콘솔에서 메시지를 입력하는 즉시, 다른 터미널에 띄워둔 컨슈머 콘솔에 해당 메시지가 그대로 출력되는 것을 확인할 수 있다. 참고로 여기서는 프로듀서가 --broker-list, 컨슈머가 --bootstrap-server 옵션을 쓰는데, 최신 Kafka CLI에서는 두 콘솔 도구 모두 --bootstrap-server로 통일해서 쓸 수 있다.

4.2 우분투에 직접 설치

Kafka는 자바 기반이라 JDK 설치가 먼저 필요하다.

  
sudo apt update
sudo apt install openjdk-17-jdk
java -version

Kafka 다운로드 및 배치

wget https://archive.apache.org/dist/kafka/3.6.0/kafka_2.13-3.6.0.tgz
tar xvf kafka_2.13-3.6.0.tgz
sudo mv kafka_2.13-3.6.0 /opt/kafka

환경 변수를 등록하고 바로 적용한다.

  
export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin

source ~/.bashrc

힙 메모리 설정

EC2 같은 클라우드 인스턴스는 메모리가 넉넉하지 않은 경우가 많다. Kafka와 주키퍼를 함께 띄우는 데 필요한 메모리가 대략 1.5G 정도인데, 흔히 쓰는 1G짜리 인스턴스에서는 기본 설정으로 구동이 안 될 수 있어 힙 크기를 줄여줘야 한다.

  
nano ~/.bashrc
# 아래 한 줄 추가
export KAFKA_HEAP_OPTS="-Xmx400m -Xms400m"

source ~/.bashrc
echo $KAFKA_HEAP_OPTS

주키퍼 실행

zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties
jps -vm

Kafka 서버 설정 및 실행

config/server.properties에서 리스너 정보와 토픽 관련 옵션을 확인한다.

  
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://127.0.0.1:9092 # 실제 ip를 적으면 외부에서 접근 가능

delete.topic.enable=true
auto.create.topics.enable=true

각 설정값의 의미는 다음과 같다.

설정	의미
`listeners=PLAINTEXT://:9092`	브로커가 어떤 프로토콜과 포트로 요청을 받을지 지정. `PLAINTEXT`는 암호화 없는 일반 통신을 뜻하고, 포트 앞이 비어있으면(`://:9092`) 모든 네트워크 인터페이스에서 `9092` 포트로 요청을 받는다
`advertised.listeners=PLAINTEXT://127.0.0.1:9092`	클라이언트(프로듀서/컨슈머)에게 “나한테 접속하려면 이 주소로 와라”라고 알려주는 주소. `listeners`는 브로커가 실제로 바인딩하는 주소, `advertised.listeners`는 외부에 공지하는 주소라서 서로 다를 수 있다
`delete.topic.enable=true`	`kafka-topics.sh --delete` 명령어로 토픽 삭제를 허용할지 여부
`auto.create.topics.enable=true`	존재하지 않는 토픽에 프로듀서가 메시지를 보내거나 컨슈머가 구독을 시도할 때, 그 토픽을 자동으로 만들어줄지 여부

특히 advertised.listeners가 중요한데, 127.0.0.1(로컬호스트)로 되어 있으면 같은 컴퓨터에서만 접속할 수 있다. 이 값을 실제 서버의 IP(예: EC2 퍼블릭 IP)로 바꿔줘야 외부에 있는 다른 컴퓨터에서도 이 브로커에 접속할 수 있다.

설정을 저장한 뒤 브로커를 실행한다.

kafka-server-start.sh -daemon /opt/kafka/config/server.properties
jps -m

5. 주요 개념

개념	설명
Zookeeper	브로커 정보를 관리하는 분산 코디네이터. 최신 버전에서는 Kafka 자체 관리(KRaft)로 대체되어 없어도 되는 방향으로 가고 있다
Topic	데이터를 구분하는 논리적 단위로, 하나 이상의 파티션을 가진다. 파티션에 쌓이는 데이터를 레코드라고 부른다
Partition	Kafka 병렬 처리의 핵심 단위. 컨슈머 그룹이 파티션 단위로 레코드를 나눠 처리하며, 컨슈머와 파티션 수를 함께 늘리면 처리량도 늘어난다. 내부 동작은 큐 자료구조와 비슷하다

6. 파이썬에서 Kafka 사용

kafka-python 패키지를 설치해서 프로듀서와 컨슈머를 각각 만들 수 있다.

pip install kafka-python

프로듀서

  
from kafka import KafkaProducer
import json

class MessageProducer:
    def __init__(self, broker, topic):
        self.broker = broker
        self.topic = topic
        self.producer = KafkaProducer(
            bootstrap_servers=self.broker,
            value_serializer=lambda x: json.dumps(x).encode("utf-8"),
            acks=0,
            api_version=(2, 5, 0),
            key_serializer=str.encode,
            retries=3
        )

    def send_message(self, msg, auto_close=True):
        try:
            # send()는 일단 내부 버퍼에 쌓인다
            future = self.producer.send(self.topic, value=msg, key="key")
            # 실제 전송은 버퍼가 차거나 flush()를 호출해야 일어난다
            self.producer.flush()
            if auto_close:
                self.producer.close()
            future.get(timeout=2)  # 비동기 전송이라 결과를 기다려줘야 한다
            return {"status_code": 200, "error": None}
        except Exception as exc:
            raise exc

broker = ["localhost:9092"]
topic = "exam-topic"
pd = MessageProducer(broker, topic)

msg = {"name": "adam", "age": 50}
res = pd.send_message(msg)
print(res)

컨슈머

  
from kafka import KafkaConsumer
import json

class MessageConsumer:
    def __init__(self, broker, topic):
        self.broker = broker
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers=self.broker,
            value_deserializer=lambda x: x.decode("utf-8"),
            group_id="my-group",
            auto_offset_reset="earliest",
            enable_auto_commit=True
        )

    def receive_message(self):
        try:
            for message in self.consumer:
                print(message.value)
        except Exception as exc:
            raise exc

broker = ["localhost:9092"]
topic = "exam-topic"

cs = MessageConsumer(broker, topic)
cs.receive_message()

7. 실무 사용 예시

Kafka를 도입한다고 해서 애플리케이션 간의 모든 통신이 Kafka를 거치는 것은 아니다. 실무에서는 보통 즉시 응답이 필요한 요청은 기존 방식(REST API, gRPC 등)으로 처리하고, 상태 변경 사실을 여러 곳에 전파해야 하는 부분만 Kafka로 넘기는 하이브리드 구조로 쓰인다.

클라이언트 → (동기 API 호출) → 주문 서비스
                                    │
                                    └─ (Kafka에 "주문 생성됨" 이벤트 발행)
                                              │
                          ┌───────────────────┼───────────────────┐
                          ▼                   ▼                   ▼
                     결제 서비스           재고 서비스           알림 서비스
                    (이벤트 구독)         (이벤트 구독)         (이벤트 구독)

주문 생성처럼 즉각적인 성공/실패 응답이 필요한 부분은 API로 처리하고, 그 이후 “주문이 생성됐다”는 사실을 여러 서비스에 전파하는 부분만 Kafka가 담당하는 식이다. 이런 구조가 실제로 쓰이는 대표적인 예시는 다음과 같다.

마이크로서비스 간 이벤트 전파: 하나의 서비스에서 발생한 상태 변경을 여러 서비스가 각자의 관심사에 맞게 구독해서 처리한다. 서비스끼리 서로의 존재를 몰라도 되므로 결합도가 낮아진다.
CQRS / 이벤트 소싱: Command(쓰기) 쪽에서 상태가 바뀔 때마다 이벤트를 Kafka에 발행하고, Query(읽기) 쪽 모델은 그 이벤트를 구독해서 조회에 최적화된 형태로 데이터를 갱신한다.
로그/이벤트 수집 파이프라인: 여러 서버에서 발생하는 로그나 사용자 행동 이벤트를 한 곳(Kafka)에 모은 뒤, 이를 필요로 하는 분석 시스템·모니터링 시스템·데이터 레이크로 각각 흘려보낸다.
실시간 스트림 처리: Kafka Streams나 Spark Streaming 같은 도구와 연결해서, 들어오는 데이터를 실시간으로 집계·변환하는 파이프라인을 구성한다.
알림 팬아웃(fan-out): 하나의 이벤트(예: 결제 완료)를 이메일, 푸시 알림, SMS 등 서로 다른 채널을 담당하는 서비스들이 동시에 구독해서 각자 처리한다.

공통적으로 “요청-응답”이 아니라 “발생한 사실을 여러 소비자가 비동기적으로 나눠 처리한다”는 패턴에 Kafka가 잘 들어맞는다.

8. 정리

Kafka는 애플리케이션끼리 직접 연결되던 구조를 중앙 집중형 파이프라인으로 바꿔주는 메시지 브로커로, 높은 처리량과 확장성, 영속성, 고가용성을 무기로 데이터 파이프라인의 표준처럼 자리 잡았다. 프로듀서가 토픽에 데이터를 보내면 파티션 단위로 저장되고, 컨슈머(그룹)가 이를 병렬로 읽어가는 구조가 핵심이며, Docker든 우분투 직접 설치든 결국 주키퍼(또는 KRaft)와 브로커를 띄우는 과정이라는 점은 동일하다.

이번 글에서는 Kafka 자체의 개념과 설치, 파이썬 클라이언트로 메시지를 주고받는 방법까지 다뤘다.

Kafka