Spring Boot - Kafka!

작성일 Edited on In

개요

https://kafka.apache.org/

출처: https://github.com/AndersonChoi/tacademy-kafka

카프카는 과거 링크드인의 제이 크렙스, 준 라오, 네하 나크헤데에 의해 탄생한 스칼라로 개발한 오픈 소스 메시지 브로커 프로젝트이다

기존의 end to end 방식의 연결 구조에서 messaging broker 방식으로 변경하기 위한 메시지 브로커 오픈소스
기존의 복잡한 파이프라인 구조를 아래와 같이 kafka를 통해 간편화시킨다.

kafka1

kafka1

카프카를 사용함으로 얻을수 있는 장점은 아래와 같다.

  • 높은 처리량
  • 순서 보장
  • 적어도 한 번 (at least once) 전송 방식
  • 강력한 파티셔닝
  • 자연스러운 백 프레셔 핸들링
  • 로그 컴팩션

이번 포스트에선 스프링 부트에서 카프카를 사용하는 방법을 알아보는 것임으로 카프카의 간단한 구성요소만 알아보고 바로 라이브러리 사용법을 살펴볼 예정이다.

용어

  • Broker : 카프카 애플리케이션 서버 단위
  • Topic : 데이터 분리 단위. 다수 파티션 보유
  • Partition : 레코드 저장소. 컨슈머 요청시 레코드 전달
  • Offset : 각 레코드당 파티션에 할당된 레코드 고유 번호
  • Consumer : 레코드를 가져가는(polling) 애플리케이션
  • Consumer group : 다수 컨슈머 묶음
  • Consumer offset : 특정 컨슈머가 가져간 레코드의 번호
  • Producer : 레코드를 브로커로 전송하는 레코드 저장 애플리케이션

파티션(메시지 저장소)의 복제(고가용성)를 위해 일반적으로 3개 이상의 broker를 운영하며
확장(스케일 아웃), 장애 발생 시 메시지 백업 등의 안정성을 제공한다.

파티션

다른 브로커와는 다르게 파티션이라는 저장공간을 제공하며 저장되는 리소스(메시지)의 백업, 복제 등을 제공한다.

kafka3

파티션을 나눔으로 여러개의 클라이언트(consumer) 가 각 파티션에 연결하여 동시에 처리할 수 있다.

파티션 개수마다 consumer 를 사용해야한다. 물론 하나의 consumer 가 여러개의 파티션을 담당할 수 있다.

파티션과 consumer 가 N:1 관계를 맺기에 consumer 끼리 경쟁할 필요가 없으며
어떤 consumer 가 데이터를 어디까지 가져갔는지 파악가능하다.

트래픽이 많아질수록 파티션과 consumer 를 늘리는 전략으로 스케일 아웃이 가능하다.
단 파티션은 한번 늘리면 줄이는 것은 불가능함으로 파티션을 늘리기전 충분한 테스트 과정이 필요하다.

반대로 파티션 개수보다 consumer 가 많아지면 추가생성된 consumer는 동작하지 않는다.

복제

여러개의 브로커를 구성하고 하나의 토픽에 여러개의 파티션을 지정하면 브로커별로 파티션을 나누어 저장한다.
아래 그림의 경우 3개의 브로커에 3개의 파티션을 지정한 경우이다.

kafka6

모종의 이유로 셋 중 하나의 브로커가 중단된다면 데이터 손실이 발생하므로 카프카에서는 아래와 같이 복제 정책을 사용한다.

kafka7

파티션이 많아질수록 동기화에 많은 리소스(네트워크, 저장소 등)가 사용되므로 사용량에 따른 적절한 브로커 개수를 설정해야 한다.

컨슈머 그룹

다른 메시지 브로커 역할을 수행하는 애플리케이션이 그렇듯
여러 개의 인스턴스가 동일한 topic을 구독하고 있을 때 중복 처리를 방지하기 위한 그룹 설정이 필요하다.

kafka5

그룹을 지정하고 consumer 생성후 파티션에서 레코드를 읽어가면 읽은 부분까지 offset 을 kafka 서버에 기록(commit)한다.

그룹별로 offset 을 지정하기 때문에 같은 topic 에 대한 여러가지 로직을 수행하고 싶다면
해당 레코드를 처리하는 여러 consumer 그룹을 생성하면 된다.

zookeeper

주키퍼는 Yahoo에서 처음 개발되어 Apache License로 오픈소스 된 프로젝트로 로드밸런스, 시스템 간의 정보 공유, 상태 체크, 서버들 간의 동기화 등을 처리해준다.

카프카 역시 수많은 데이터를 관리하는 분산 데이터 시스템이기에 분산, 복제된 데이터의 충돌을 막고 데이터 동기화를 진행해야 하는데 이를 처리하는 분산 코디네이션 서비스가 주키퍼이다.

kafka4

초록색 박스가 주키퍼 서버역할이고 카프카 노드는 주키퍼 클라이언트 역할이다.

주키퍼는 카프카의 데이터를 액세스하여 데이터를 동기화하고 카프카 노드에 대한 메타데이터도 관리하는데
카프카가 성장하면서 zookeeper의 성능 한계가 드러나면서 향후 zookeeper는 제외되고 카프카 라프트(KRaft)가 대체할 예정이다.

설치

https://github.com/onlybooks/kafka2/blob/main/appendix_C/single_zk_kafka/docker-compose.yml

간단히 로컬에 싱글 클러스터로 설치한다. 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
version: "3.5"
services:
  zk:
    image: confluentinc/cp-zookeeper:5.5.1
    restart: always
    hostname: zk
    container_name: zk
    ports:
      - "2181:2181"
    environment:
      - ZOOKEEPER_SERVER_ID=1
      - ZOOKEEPER_CLIENT_PORT=2181
      - ZOOKEEPER_TICK_TIME=2000
      - ZOOKEEPER_INIT_LIMIT=5
      - ZOOKEEPER_SYNC_LIMIT=2
      - ZOOKEEPER_SERVERS=zk:2888:3888

  kafka:
    image: confluentinc/cp-kafka:5.5.1
    restart: always
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
      - "9999:9999"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zk:2181
      KAFKA_LISTENERS: INTERNAL://kafka:9092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9999

  kafka_manager:
    image: hlebalbau/kafka-manager:stable
    container_name: cmak
    ports:
      - "9000:9000"
    environment:
      ZK_HOSTS: "zk:2181"
      APPLICATION_SECRET: "random-secret"
    command: -Dpidfile.path=/dev/null

생성후 컨테이너 안으로 들어가 각종 kafka cli 툴들을 사용할 수 있다. 

1
2
3
4
docker exec -it kafka bash
root@kafka:/# kafka-topics --version

5.5.1-ccs (Commit:3c4783aac9e33249)