Kafka 클러스터 구축을 위해서는 zookeeper 가 필요하다. 일반적으로 kafka 클러스터 구축은 3개 이상의 broker를 가지는데 이 ‘세 개 이상의 broker가 하나처럼 동작하기 위한 정보는 zookeeper가 관리’하기 때문에다.

기본적으로 프로듀서 혹은 컨슈머는 카프카에 연결할 때 직접 카프카 브로커에 연결하는 것이 아니라 주키퍼에 연결하여 카프카 브로커, 토픽, 파티션 등의 정보를 취득한 뒤 브로커에 연결하게 된다.¹

Zookeeper 설치

zoookeeper 설치는 zookeeper #1. 개요, #2. 설치와 설정, #3. 구동과 확인 포스트를 참고하면 된다.

다운로드 및 압축 해제

Apache kafka 공식 페이지에서 바이너리를 다운로드 한다.
다운로드한 파일을 FTP 또는 SFTP등을 이용하여 서버에 업로드한다.
또는 wget 명령을 이용하여 서버에서 다운로드 한다

설정

카프카 클러스터의 설정을 위해서는 ‘zookeeper’ 클러스터가 존재하는 상태에서 $KAFKA_HOME/config/server.config 파일에 기재된 설정만 잘 조절해주면 된다.

############################# Server Basics #############################
# 브로커 고유 ID. 각각의 브로커는 중복되지 않은 고유한 숫자 값을 가진다. 
broker.id=1

############################# Socket Server Settings #############################
# Broker 가 사용하는 호스트와 포트
listeners=PLAINTEXT://dist01.haedongg.net:9092

# Producer와 Consumer가 접근하는 호스트와 포트
# 다음의 연결 옵션을 제공한다 -PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
advertised.listeners=PLAINTEXT://dist01.haedongg.net:9092

# 네트워크 요청 처리 스레드
num.network.threads=3

# IO 발생 시 생기는 스레드 수
num.io.threads=8

# 소켓 서버가 사용하는 송수신 버퍼
socket.send.buffer.bytes=1024000
socket.receive.buffer.bytes=1024000

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600

############################# Log Basics #############################
# Broker가 받은 데이터 관리를 위한 저장 공간
log.dirs=/home/kafka/data

# 여러 개의 디스크, 여러 개의 디렉토리를 사용하는 경우 콤마로 구분하여 추가한다.
# log.dirs=/home/kafka/data,/mnt/sdb1/kafka/data,/third_disk/mq-data

# 토픽 당 파티션의 수. 값 만큼 병렬 처리 가능하고, 파일 수도 늘어난다.
# 기본 값으로써 토픽을 생성할 때 파티션 숫자를 지정할 수 있고, 변경도 가능하다.
num.partitions=1

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state" 
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.

# 토픽에 설정된 replication의 인수가 지정한 값보다 크면 새로운 토픽을 생성하고 작을 경우 브로커의 숫자와 같게 된다.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################
# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################
# 수집 데이터 파일 삭제 주기. (시간, 168h=7days)
# 실제 수신된 메시지의 보존 기간을 의미한다. 
log.retention.hours=168

# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824

# 토픽별 수집 데이터 보관 파일의 크기. 파일 크기를 초과하면 새로운 파일이 생성 된다. 
log.segment.bytes=1073741824

# 수집 데이터 파일 삭제 여부 확인 주기 (밀리초, 300000ms=5min)
log.retention.check.interval.ms=300000

# 삭제 대상 수집 데이터 파일의 처리 방법. (delete=삭제, compact=불필요 내용만 제거)
log.cleanup.policy=delete

# 수집 데이터 파일 삭제를 위한 스레드 수
log.cleaner.threads=1

############################# Zookeeper #############################
# zookeeper 정보.
# zookeeper 클러스터 모두 콤마(,)로 구분해서 기재한다.
zookeeper.connect=dist01.haedongg.net:2181,dist02.haedongg.net:2181,dist03.haedongg.net:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000

############################# Group Coordinator Settings #############################
# 초기 GroupCoordinator 재조정 지연 시간 (밀리초) 
# 운영환경에서는 3초를 권장 
group.initial.rebalance.delay.ms=3000

############################# 기타 ################################
# 최대 메시지 크기 (byte) 
# 예시는 15MB 까지의 메시지를 수신할 수 있다.
message.max.bytes=15728640

구동 및 확인

Kafka broker 구동을 위해서는 JAVA가 필요하다. JDK가 설치되어있고 환경변수($JAVA_HOME 및 $PATH)가 설정되어 있다면 별도의 설정 없이 카프카 브로커를 구동할 수 있다. 만약 JDK가 설치되어있지 않거나, 별도의 JAVA를 사용하고자 한다면 export 명령을 이용하여 변수를 선언하거나, $KAFKA_HOME/bin/kafka-run-class.sh 파일에 JAVA관련 변수를 넣어주도록 한다.

# Memory options
# 다음 $JAVA_HOME 관련 옵션을 적절히 수정한다.
# JAVA_HOME=/WHERE/YOU/INSTALLED/JAVA 이렇게 JAVA_HOME 변수를 지정하면 된다. 
# 만약 JAVA_HOME 변수가 선언되지 않았다면  PATH에 등록된 경로에서 java 명령을 찾게 된다. 
if [ -z "$JAVA_HOME" ]; then
  JAVA="java"
else
  JAVA="$JAVA_HOME/bin/java"
fi

# Memory options
# 카프카 구동을 위한 java heap memory 옵션
if [ -z "$KAFKA_HEAP_OPTS" ]; then
#  KAFKA_HEAP_OPTS="-Xmx256M"   이 값이 기본
  KAFKA_HEAP_OPTS="-Xms256M -Xmx1G"
fi

broker 실행

broker로 구동할 모든 Kafka 서버에서 수행한다.

$KAFKA_HOME/kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
ex) /home/apps/kafka/bin/kafka-server-start.sh -daemon /home/apps/kafka/config/server.properties
# 만약 -daemon 옵션을 추가하지 않으면 브로커가 background가 아닌 foreground에서 구동된다.

프로세스 및 리슨 포트 확인

jps -l

18320 kafka.Kafka
18403 sun.tools.jps.Jps
17899 org.apache.zookeeper.server.quorum.QuorumPeerMain

netstat -nltp

Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name
tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN      973/sshd
tcp        0      0 127.0.0.1:25            0.0.0.0:*               LISTEN      1286/master
tcp6       0      0 :::8080                 :::*                    LISTEN      17899/java
tcp6       0      0 :::40434                :::*                    LISTEN      17899/java
tcp6       0      0 :::22                   :::*                    LISTEN      973/sshd
tcp6       0      0 ::1:25                  :::*                    LISTEN      1286/master
tcp6       0      0 :::34880                :::*                    LISTEN      18320/java
tcp6       0      0 :::9092                 :::*                    LISTEN      18320/java
tcp6       0      0 :::2181                 :::*                    LISTEN      17899/java

종료

$KAFKA_HOME/bin/kafka-server-stop.sh

2022-04-112022-05-06

Kafka #1. 개요

Zookeeper #1. 개요
Zookeeper #2. 설치와 설정
Zookeeper #3. 구동과 확인

Kafka #1. 개요
Kafka #2. 설치

아파치 카프카(Apache Kafka)는 아파치 소프트웨어 재단이 스칼라로 개발한 오픈 소스 메시지 브로커 프로젝트이다. 이 프로젝트는 실시간 데이터 피드를 관리하기 위해 통일된, 높은 처리량, 낮은 지연시간을 지닌 플랫폼을 제공하는 것이 목표이다. 요컨대 분산 트랜잭션 로그로 구성된, 상당히 확장 가능한 pub/sub 메시지 큐로 정의할 수 있으며, 스트리밍 데이터를 처리하기 위한 기업 인프라를 위한 고부가 가치 기능이다.

큐 (Queue)

먼저 큐에 대한 이해가 필요한데, 말 그대로 ‘기다리는 줄(열)’, ‘대기 열’을 생각하면 된다. 전공자라면 자료구조, 알고리즘 과목 등에서 접하게 되는데 Stack 과 함께 배운다.
FIFO¹ 구멍이 두 개인 파이프의 한쪽에 동전을 넣는다고 생각하면 된다. 계속 동전을 밀어 넣다 보면 가장 먼저 넣었던 동전부터 동전이 밀려나오는 구조가 된다.
이와 반되 되는 것이 Stack. 쌓는다는 의미의 stack으로 FILO² 컵에 동전을 쌓는 것을 상상하면 된다. 컵에 동전을 쌓아 넣다다가 동전을 꺼내면 나중에 넣었던 동전을 먼저 꺼내야 한다.

Message Queue

이렇게 먼저 넣은 값이 먼저 출력되는 구조를 이용해 메시지³를 전송하는 것이 Message Queue이다. Rabbit MQ, IBM MQ, Apache active MQ , Rocket MQ 등 다양한 종류의 Message Queue가 존재한다.
기본적인 개념은 모두 같지만 메시지를 주고 받는 방식에 따라 크게 두 종류로 구분할 수 있다.
– Push : 서버가 메시지를 클라이언트에 보내주는 방식
– Pull : 클라이언트가 메시지를 서버에서 가져오는 방식

별것 아닌 것 같지만 Pull 과 Push는 큰 차이를 가진다.

종류	Push	Pull
장점	· 메시지 발생 즉시 클라이언트에 전달	· 필요한 클라이언트만 메시지를 가져가므로 서버 부하 감소 · 네트워크 부하 감소
단점	· 서버가 모든 클라이언트에 연결해야 하므로 서버 부하 증가	· 메시지를 전달한 클라이언트 확인 불가
주 용도	카카오톡 등 메시지, 알림 전달	분석용 서버 로그 데이터 등

Push 방식과 Pull 방식 비교

Apache Kafka

Apache Kafka는 Pull 방식의 Message Queue이다.
중요한 개념으로 메시지를 발생 시키는 (또는 서버의 Queue에 메시지를 보내는) Producer(또는 Publisher)와 메시지를 가져가는 Consumer(또는 subscriber)가 있다.⁴

즉, 누군가 메시지를 보내는 곳, 누군가 메시지를 꺼내갈 곳이 Kafka 이다. ⁵

이 Kafka Broker가 하는 일과 핵심적인 개념은 다음과 같다.

Queue : 가장 핵심적인 기능이다. 수신된 메시지를 저장한다. kafka가 인지하는 형식으로 디스크에 저장된다.
Offset 관리 : 프로듀서가 보낸 메시지와 컨슈머가 가져간 메시지의 offset, 즉, 몇 개의 메시지가 수신되었고 어떤 클라이언트가 어디서부터 어디까지의 메시지를 가져갔는지 기억하고 관리한다.
Topic : Topic은 앞서 설명한 메시지를 담는 파이프 하나를 떠올리면 된다. 메시지를 넣을 파이프를 만들고 이름을 붙여서 관리한다. 이 Topic은 여러개의 Pratition으로 나뉠 수 있다.
Replica : 데이터의 복제, Broker나 디스크의 장애 등에 대비해 데이터를 복제하는 기능이다.

Public cloud에서

AWS : MKS(Managed Kafka Service)
GCP : Pub/Sub (Publisher와 Subscriber)
AZURE: HDInsight Kafka

2020-12-292020-12-29

Openstack #1-4 MessageQueue(RabbitMQ)

Openstack #1 개요
Openstack #1-1 선행작업
Openstack #1-2 MariaDB(MySQL) 설치
Openstack #1-3 Chrony 설정
Openstack #1-4 MessageQueue(RabbitMQ)
Openstack #1-5 Memcached
Openstack #1-6 Etcd

RabbitMQ는 오픈 소스 메시지 브로커 소프트웨어(메시지 지향 미들웨어)로서, AMQP를 구현하였으며 그 이후로 STOMP ,MQTT 등의 프로토콜을 지원하기 위해 플러그인 구조와 함께 확장되고 있다. 메시지를 생산하는 생산자(Producer)가 메시지를 큐에 저장해 두면, 메시지를 수신하는 소비자(Consumer)가 메시지를 가져와 처리하는 Publish/Subscribe 방식의 메시지 전달 브로커이다.

rabbitmq 서버는 controller 노드에만 설치한다.

패키지 설치

[root@openstack1:~]#yum -y install rabbitmq-server
Loaded plugins: fastestmirror
Loading mirror speeds from cached hostfile
 * base: mirror.kakao.com
Package rabbitmq-server-3.6.16-1.el7.noarch already installed and latest version
Nothing to do

서비스 등록 및 구동

[root@openstack1:~]#systemctl enable rabbitmq-server.service
[root@openstack1:~]#systemctl start rabbitmq-server.service
 # 또는
[root@openstack1:~]#chkconfig rabbitmq-server on
알림: 'systemctl enable rabbitmq-server.service'에 요청을 전송하고 있습니다.
[root@openstack1:~]#service rabbitmq-server start
Redirecting to /bin/systemctl start rabbitmq-server.service

사용자(계정) 생성

# rabbitmqctl add_user ACCOUNT PASSWORD
[root@openstack1:~]# rabbitmqctl add_user openstack openstack
Creating user "openstack"
 # 기존 사용자를 삭제하려면 add_user 대신 delete_user명령을 사용하면 된다.

권한 부여

[root@openstack1:~]#rabbitmqctl set_permissions openstack ".*" ".*" ".*"
Setting permissions for user "openstack" in vhost "/"