Kafka

Apache 만든 메시지 큐로 대용량 실시간 로그 처리에 특화되어 설계 되어 있으며, 메시지를 메모리에 저장하는 기존 메시지 큐 시스템과 다르게 Kafka는 파일 시스템에 저장하여 파일의 유실 우려가 적음.

또한 기존의 메시지 큐 시스템은 broker가 consumer에게 메시지를 push 해주는 방식이지만 Kafka는 Consumer가 Broker로부터 직접 가져오는 pull 방식이어서 (pub/sub 모델) Cosumer가 처리 할 수 있는 범위 내에서만 메시지를 처리할 수 있게 설계할 수 있다.

Kafka 구성 요소 및 특징

Topic, Partition

카프카에서 메시지는 Topic으로 분류가 된다. Topic은 여러개의 Partition으로 나뉘어 진다.

Partition내에서 가장 마지막 offset 뒤에 메시지가 저장되므로, Partition내에서는 순서가 보장이 되지만 실제로는 Consumer의 동작 방식에 의해 순서가 보장 되지 않는다.

Round-robin 방식으로 메시지가 써지게 되고 메시지의 순서를 보장해 주지 않기 때문에 주의하여야 한다.

한 번 파티션의 수를 늘리면 줄일 수 없다.

Log

Partition의 한 칸에 해당하는 메시지로 Key,value, Timestamp로 구성 된다.

Offset

Partition내에서 각 메시지를 식별할 수 있는 Uique id

메시지를 소비하는 Consumer가 읽을 차례를 의미하고, Partition마다 별도로 관리 된다.

Producer

메시지를 생산하는 주체로, 정해진 Topic으로 메시지를 기록한다.

Consumer

메시지를 소비하는 주체로, Topic을 구독함으로써, 해당 topic내의 각 Partition에 존재하는 offset의 위치를 통해 이전에 소비했던 offset을 기억하고 관리하기 때문에 Consumer가 죽었다가 다시 살아나도 전에 마지막으로 읽었던 위치에서부터 다시 읽어들일 수 있다.

Consumer Group

Consumer group은 한 개의 Topic을 담당한다. (하나의 토픽은 여러 Goup이 접근할 수 있다.)

Topic내 Partition에서 다음에 소비할 offset이 어디인지 공유하면서 메시지를 소비한다.

일반적으로 Partition수와 Consumer는 1:1 매칭이 이상적이다.

• Consumer Group이 필요한 이유

  Group내의 Consumer가 Down되면 메시지를 소비할 수 없게 된다. 이를 Rebalance라고 하는데, Consumer Group은 Partition의 Offset을 공유하여 다른 Consumer가 소비를 계속 할 수 있게 해준다.

Broker

Kafka 서버, 한 노드 내에 여러개의 Broker를 띄울 수 있다.

Zookeeper

Zookeeper의 용도는 클러스터 최신 설정정보 관리, 동기화, 리더 채택 등 클러스터 서버들이 공유하는 데이터를 관리하기 위해 사용한다.

Broker에 분산 처리된 메시지 큐의 정보들을 관리하고 Zookeeper없이는 Kafka 구동이 불가능하다.

설치하기

Docker compose를 활용하여 Zookeeper와 Kafka를 한번에 설치해준다.

version: '3.5'
services:
  zookeeper:
    container_name: zookeeper-container
    image: 'bitnami/zookeeper:latest'
    ports:
    - '2181:2181'
    environment:
    - ALLOW_ANONYMOUS_LOGIN=yes
    - ZOO_MY_ID=1
  zookeeper2:
    container_name: zookeeper-container2
    image: 'bitnami/zookeeper:latest'
    ports:
    - '2182:2181'
    environment:
    - ALLOW_ANONYMOUS_LOGIN=yes
    - ZOO_MY_ID=2
  kafka:
    container_name: kafka-container
    image: 'bitnami/kafka:latest'
    depends_on:
      - zookeeper
    ports:
    - '9092:9092'
    environment:
    - KAFKA_ADVERTISED_HOST_NAME=127.0.0.1
    - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181,zookeeper2:2182
    - ALLOW_PLAINTEXT_LISTENER=yes

$ docker-compose up

간단 테스트

카프카 동작 확인을 위해 컨테이너 내부에 접속한다.

$ docker exec -it kafka-container /bin/bash

• Topic 생성

  $ kafka-topics.sh --create -bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic "test"
  

  • bootstrap-server : 브로커 리스트
  • replication-factor : partion 복제 수, 기본 값은 파티션 복제를 사용하지 않음
  • partitions : 토픽이 생성되거나 변경될 때의 파티션 수
  • topic : create, alter, describe, delete 옵션에 사용할 토픽 명

  파티션이 늘릴 수는 있는데 줄일 수는 없으니 늘릴 때 유의

• Topic 목록

  $ kafka-topics.sh --list --bootstrap-server localhost:9092
  

• Topic 삭제

  $ kafka-topics.sh --delete --topic "test" --bootstrap-server localhost:9092
  

• Topic 상세 보기

  $ kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic "test"
  

• Topic 설정 변경

  Topic 보관 주기를 1시간으로 변경 (기본값 7일)

  $ kafka-topics.sh --zookeeper localhost:2181 --alter --topic "test" --config retention.ms=3600000
  

• Topic의 Partition 수 변경

  $ kafka-topics.sh --zookeeper localhost:2181 --alter --topic "test" -partitions 2
  

• Producer

  $ kafka-console-producer.sh --broker-list localhost:9092 --topic "test"
  

• Consumer

  $ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic "test" --from-beginning
  

  –from-beginning : Consumer에게 설정된 offset이 없으므로 가장 최신의 메시지 대신 가장 먼저 도착한 메시지부터 읽도록 하는 옵션

• Consumer group create

  $ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --group my-group
  

• Consumer group list

  $ kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
  

• Consumer 상태, 오프셋 확인

  $ kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --describe
  

–broker-list : kafka 서버 목록 (producer는 zookeeper에 접속할 필요 없이 브로커에 메시지를 넘겨주면 됨)

–bootstrap-server : kafka 서버 목록 (broker-list와 동일하지만 consumer는 zookeeper에 접속하여야 함)

–zookeeper (topic이 사용)

• kafka 0.8 이전의 경우 partition의 offset이 zookeeper로 기록되어서 consumer는 zookeeper를 알아야 했지만 이 후에는 broker에 기록이 되어 bootstrap-server를 사용
• 만약 구버전을 써서 zookeeper에 offset이 저장된다면 (offsets.storage=zookeeper) –zookeeper 를 사용한다.

• 임의의 트래픽 발생시키기

$ kafka-producer-perf-test.sh --num-records 1000 --throughput 100 --record-size 100 --print-metric --topic "test" --producer-props bootstrap.servers=localhost:9092

–num-records: 전송할 메시지 수

–throughput: 초당 메시지 처리량

–record-size: 메시지 사이즈 (랜덤문자열인가?)

–print-metric: 테스트 종료 후 metric출력

–producer-props: producer 설정 값 (bootstrap.servers 필수)

추가

bitnami/kafka의 설정 파일은 /opt/bitnami/kafka/config/에 있다.

Kafka API Document