Kafka 설치 방법과 기본 명령어를 알아보겠습니다.
실제 사용은 꽤 복잡하지만, 설치나 기본 명령어는 간단합니다.

설치

두 가지 방법이 있습니다.
먼저 직접 다운로드 하는 방법입니다.

Kafka 공식 사이트에서 다운 받아도 되고, wget 명령어를 활용해도 됩니다.
설치하고 싶은 디렉토리에서 아래 명령어를 실행하고 압축을 풀어줍니다.

wget  https://archive.apache.org/dist/kafka/2.8.1/kafka_2.12-2.8.1.tgz

tar -xzf kafka_2.12-2.8.1.tgz

다음으로 Homebrew를 사용한다면, 간단하게 설치할 수 있습니다.
아래 명령어를 실행하고, /homebrew/bin/ 경로에 가면, Kafka 디렉토리를 볼 수 있습니다.

brew install kafka

이후 별도 추가 설치 과정은 없습니다.

기본 명령어

모든 명령어는 Kafka 루트 폴더를 기준으로 합니다.

# Version 확인
./bin/kafka-topics.sh --version

# Zookeeper 실행
./bin/zookeeper-server-start.sh ./config/zookeeper.properties

# Kafka 실행
./bin/kafka-server-start.sh ./config/server.properties

# Topic 생성
./bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic stock-update

# Topic 생성 - partitions, replication-factor
./bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic stock-update --partitions 1 --replication-factor 1

# Topic 확인
./bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic stock-update

# Topic 목록
./bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# Topic 삭제
./bin/kafka-topics.sh --delete --bootstrap-server localhost:9092 --topic stock-update

# Producer
./bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic stock-update

# Producer - Key, Value
./bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic stock-update --property "parse.key=true" --property "key.separator=:" --property "print.key=ture"

# Producer - Message
ABCDE

# Producer - Key, Value Message
key:{"val1":"A","val2":"B","val3":3}

# Consumer
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic stock-update --from-beginning

# Consumer Group 확인
./bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

# Consumer Group Topic 확인
./bin/kafka-consumer-groups.sh --describe --bootstrap-server localhost:9092 --group stock-update-group

실행 및 테스트

간단하게 로컬에서 Kafka 서버를 실행하고 테스트 해보겠습니다.
별도 코드 작성없이 제공되는 Shell Script를 활용하겠습니다.

테스트를 위해서는 총 4개의 Terminer 창을 띄워야 합니다.
먼저 Coordinator 서버인 Zookeeper를 실행합니다.

참고로 Zookeeper는 Kafka 실행을 위해 꼭 필요한 서버입니다.
하지만 Zookeeper에 대한 의존성이 너무 커지자, Kafka 3.0 부터 Zookeeper를 제거하는 기능들이 나왔습니다.

./bin/zookeeper-server-start.sh ./config/zookeeper.properties

다음으로 Kafka 서버를 실행합니다.

./bin/kafka-server-start.sh ./config/server.properties

이제 Message를 저장하는 공간인 Topic을 생성해야 합니다.

./bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic stock-update

다음으로 Message를 생산하는 Producer를 실행합니다.
Producer를 실행하면 종료 전까지 계속해서 Message를 생산할 수 있습니다.

Message 간에 구분은 Enter입니다.

./bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic stock-update
>123
>456

마지막으로 Message를 소비하는 Consumer를 실행해 줍니다.
Producer에서 Message가 생산될 때마다, 스트리밍으로 Message 정보가 출력됩니다.

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic stock-update --from-beginning