3. 카프카 설치

카프카 설치

카프카 배포판인 컨플루언트 플랫폼을 사용한 카프카 클러스터 구축 방법 소개 1. 서버를 1대만 사용하여 구성 2. 서버를 여러대 사용하여 구성

카프카 클러스터 환경 구축하기

배포판 : 컨플루언트, 클라우데라, 호튼웍스 등
배포판별 기능차이 존재. 필요한거에 맞게 선택

카프카 구축

OS 설치 : Linux, Max 만 지원
- 일부 루트권한 필요
- 여러대 사용 -> /etc/hosts 네트워크 설정 -> 서버간 서로의 호스트명으로 이름 확인

JDK : 1.8 지원

다운로드 (1.8-202버전 이하로, 퍼블리 무료버전)
- https://www.oracle.com/java/technologies/javase-downloads.html
- https://www.oracle.com/java/technologies/javase/javase8-archive-downloads.html
- 다운로드 파일 -> /tmp 에 위치시킴

CentOS

# JDK 설치
$ sudo rpm -ivh /tmp/jdk-8u202-linux-x64.rpm

# 환경변수 -> /etc/profile.d/java.sh 생성
$ sudo vi /etc/profile.d/java.sh
expert JAVA_HOME=/usr/java/default
expert PATH=$PATH:$JAVA_HOME/bin

# 반영
$ source /etc/profile.d/java.sh

# 확인
$ echo $JAVA_HOME
$ java -version

컨플루언트 플랫폼의 YUM 리포지터리 등록

  # 컨플루언트 제공 공개키 등록
  $ sudo rpm --import https://packages.confluent.io/rpm/5.0/archive.key

  # 파일작성
  $ sudo vi /etc/yum.repos.d/confluent.repo
  [Confluent.dist]
  name=Confluent repository (dist)
  baseurl=https://packages.confluent.io/rpm/5.0/7
  gpgmeck=1
  gpgkey=https://packages.confluent.io/rpm/5.0/archive.key
  enabled=1

  [Confluent]
  name=Confluent repository
  baseurl=https://packages.confluent.io/rpm/5.0
  gpgmeck=1
  gpgkey=https://packages.confluent.io/rpm/5.0/archive.key
  enabled=1

  # 기존캐시 삭제
  $ yum clean all

  # yum 사용가능한 패키지 목록 확인
  $ yum list | grep confluent

카프카 설치
컨플루언트 플랫폼 -> 잘게 구분된 여러 패키지 존재함. 그중 confluent-platform-oss-2.11은 OSS 버전 패키지 설치, 컨플루언트 문서 참고
```
# 카프카 실행을 위한 패키지 설치
$ sudo yum install confluent-platform-oss-2.11
```

브로커 데이터 디렉터리 설정 Oralce JDK 사용시 데이터 디렉터리 변경 필요 컨플루언트 플랫폼 스크립트 -> 사용자 : cp-kafka

# 브로커 디렉터리 설정
$ sudo vi /etc/kafka/server.properties
log.dirs=/var/lib/kafka/data # 로 수정, 기본 : /var/lib/kafka

# 데이터 디렉터리 추가
$ sudo mkdir /var/lib/kafka/data

# 컨플루언트 플랫폼 사용시 유저정보 권한
$ sudo chown cp-kafka:confluent /var/lib/kafka/data

여러대 동작하기 -> 추가 설정 주키퍼, 카프카클러스터 -> 여러 대의 클러스로 동작시키기 위해서는 추가설정 필요
```
# 주키퍼 추가 설정 -> 아래 내용 추가
$ vi /etc/kafka/zookeeper.properties

# 주키퍼 클러스터 초기 접속 및 동기 타임아웃
# tickTime이라는 파라미터를 단위로 계산 -> 기본값이 3000ms(3초)
initLimit=10  # 30초(10 * 3초)
syncLimit=5   # 15초(5 * 3초)

# 클러스터를 구성하는 서버군의 정보
server.1=kafka-broker01:2888:3888
server.2=kafka-broker02:2888:3888
server.3=kafka-broker03:2888:3888
```
서버 정보 형식 : server.=<서버 호스트명>:<서버 통신용 포트1>:<서버 통신용 포트2> myid : 주키퍼 클러스터 내, 각 서버에 고유하게 부여된 서버 번호 호스트 kafka-broker01 -> 1, kafka-broker02 -> 2, kafka-broker03 -> 3 번에 해당 각 myid -> /var/lib/zookeeper/myid에 작성해야 함 myid는 서버마다 다르기 떄문에 실행 명령도 서버마다 다름

각 서버별 명령어 실행

# kafka-broker01(myid=1)
$ echo 1 | sudo -u cp-kafka tee -a /var/lib/zookeeper/myid

# kafka-broker02(myid=2)
$ echo 2 | sudo -u cp-kafka tee -a /var/lib/zookeeper/myid

# kafka-broker03(myid=3)
$ echo 3 | sudo -u cp-kafka tee -a /var/lib/zookeeper/myid

브로커 추가 설정
```
$ vi /etc/kafka/server.properties
broker.id=<서버별로 정한 Broker ID> # 이미 작성된 내용 수정
broker.id.generation.enable=false # 추가
zookeeper.connect=kafka-broker01:2181,kafka-broker02:2181,kafka-broker03:2181 # 작성된 내용 수정
```
log.dirs를 변경한 상태(/var/lib/kafka -> /var/lib/kafka/data), 여기에서 그 파일에 추가로 여러 대의 서버에서 실행하기 위한 설정을 해야함
- broker.id : 브로커 ID 설정하기 위한 항목
  broker.id 설정하지 않고, broker.id.generation.enable=true 설정시 -> 브로커 ID 자동 부여
  broker.id.generation.enable=false 설정 -> broker.id 수동으로 설정
- 브로커도 주키퍼의 myid와 마찬가지로 브로커마다 고유 id부여 필요
- 브로커ID는 정수값으로 설정 해야함
- 주키퍼의 myid와 동일할 필요는 없지만 여기에서 아래와 같이 설정
  kafka-broker01 -> 1
  kafka-broker02 -> 2
  kafka-broker03 -> 3

카프카 실행 및 동작

카프카 클러스터 실행

실행 순서 : 주키퍼 실행 -> 브로커 실행

주키퍼 실행
카프카 클러스터 구축하고 있는 경우, 주키퍼가 설치된 모든 머신 실행. 순서는 상관없음
```
# 주키퍼 실행명령
$ sudo systemctl start confluent-zookeeper
```
브로커 실행
카프카 클러스터 구축하고 있는 경우, 브로커가 설치된 모든 머신 실행. 순서 상관없음
```
# 브로커 실행
$ sudo systemctl start confluent-kafka
```
출력 로그
- 주키퍼 : /var/log/kafka/zookeeper.out
- 브로커 : /var/log/kafka/server.log

카프카 클러스터 동작 확인

카프카 빌트인 도구로 메시지를 전송하고, 제대로 송수신하는지 여부 확인이 가능함

Kafka Console Producer
Kafka Console Consumer

동작확인

메시지 송수신을 위한 토픽 작성 필요
카프카 클라이언트에서 명령 실행
first-test라는 토픽 작성
주키퍼의 접속 정보가 다를 수 있기 때문에 구성한 서버에 따라 명령 다름

서버 3대로 클러스터 구축한 경우

--zookeeper : 주키퍼의 접속정보 지정
- <호스트명>:<접속포트번호> -> 예) kafka-broker01:2181 ..
- 1대의 경우 -> --zookeeper 호스트명:2181
  - /etc/kafka/server.properties 의 zookeeper.connect 필드값 확인
--create
- 토픽작성
- 토픽목록 확인 : --list, 삭제 : --delete
--topic
- 작성하는 토픽이름 지정
- 예제는 first-test
- 토픽이름 : 밑줄(_), 마침표(.) 사용금지
--partitions
- 토픽의 파티션수
- 예제는 3
--replication-factor
- 토픽의 레플리카 복제본 수
- 카프카 클러스터의 브로커 수 이하여야한다
- 3대 -> 3이하, 1대 -> 1이하 (--replication-factor 3 or 1)

input :

# 여러대 설정
$ kafka-topics --zookeeper kafka-broker01:2181,kafka-broker02:2181,kafka-broker02:2181 --create --topic first-test --partitions 3 --replication-factor 3

output :
```
Created topic "first-test"
```

토픽이 정상 생성 확인

주키퍼 접속 정보 확인 필수

# 토픽 생성 확인
$ kafka-topics --zookeeper kafka-broker01:2181,kafka-broker02:2181,kafka-broker02:2181 --describe --topic first-test

--create -> --descirbe으로 지정하여 명령. --descirbe은 토픽에 대한 상세정보 표시 옵션 Topic, PartitionCount, ReplicationFactor 표시된다.

output :
```
Topic:first-test    PartitionCount:3    ReplicationFactor:1    Configs:
Topic: first-test    Partition: 0    Leader: 1    Replicas: 1    Isr: 1
Topic: first-test    Partition: 1    Leader: 1    Replicas: 1    Isr: 1
Topic: first-test    Partition: 2    Leader: 1    Replicas: 1    Isr: 1
```
- Leader : 각 파티션의 현재 Leader Replicark 어떤 브로커에 존재하고 있는지 표시. 번호->브로커 ID
- Replicas : 각 파티션의 레플리카 보유하고 있는 브로커 목록
- Isr : In-Sync-Replicas의 약자. 복제본 중 Leader Replicas와 올바르게 동기가 실행된 복제본을 보유하고 있는 브로커 목록 표시. 동기화가 실행되지 않은 레플리카는 Isr에 포함되지 않음. Leader Replicas 자신은 Isr에 포함됨

메시지 송수신 확인

카프카 빌트인 도구인 Kafka Console Producer 사용 하여 콘솔에 데이터 입력->메시지 카프카 전송 카프카 빌트인 도구인 Kafka Console Consumer 사용 하여 토픽에서 메시지 취득

프로듀서 서버
```
# 전송 명령어 실행
$ kafka-console-pruducer --broker-list kafka-broker01:9092, kafka-broker02:9092, kafka-broker03:9092  --topic first-test
```
- --broker-list
  - 메시지를 보내는 카프카 클러스터의 브로커 호스트명,포트번호 지정
  - 호스트명:포트번호 형태로 지정
  - 여러대 -> ,(쉼표)로 연결
  - 카프카가 통신에 사용하는 기본포트 9092 사용
- --topic
  - 메시지 송신처 -> 토픽 지정
컨슈머 서버
```
# 메시지 취득 명령어 실행
$ kafka-console-consumer --bootstrap-server kafka-broker01:9092,kafka-broker02:9092,kafka-broker03:9092 --topic first-test
```
- --bootstrap-server
  - 메시지 수신하는 카프카 클러스터의 브로커 호스트명,포트번호 지정
- --topic
  - 메시지를 수신하는 토픽

카프카 클러스터 종료

실행순서 역순, 주키퍼 종료 -> 브로커 종료

# kafak (브로커) 종료
$ sudo systemctl stop confluent-kafka

# 주키퍼 종료
$ sudo systemctl stop confluent-zookeeper

Previous2. 카프카 기초 Next4. Java API를 이용하여 애플리케이션 만들기

Last updated 4 years ago