Apache Kafka

kafka producer와 cunsumer를 테스트해보자. 일단 kafka를 먼저 설치 하자. kafka를 다운받고 간단하게 producer, cunsumer 실행하고 압축을 풀고 해당 kafka 디렉토리로 이동하자. 아래와 같이 폴더들이 있다.

~/kafka.2.12-2.3.0 > ls

LICENSE   NOTICE    bin       config    libs      site-docs

아래와 같이 입력하여 zookeeper를 실행하자.

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

 

실행되면 아래와 같이 메세지가 출력이 된다.

INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)

 

이제 카프카를 실행하자.

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

 

정상실행 되면 아래와 같이 메세지가 출력 될 것이다.

 

[2019-06-28 10:26:07,736] INFO Kafka version: 2.3.0 (org.apache.kafka.common.utils.AppInfoParser)
[2019-06-28 10:26:07,736] INFO Kafka commitId: fc1aaa116b661c8a (org.apache.kafka.common.utils.AppInfoParser)
[2019-06-28 10:26:07,737] INFO Kafka startTimeMs: 1561685167728 (org.apache.kafka.common.utils.AppInfoParser)
[2019-06-28 10:26:07,740] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

 

이제 토픽을 만들자. 토픽은 메시지 저장소라 생각하면 될 것 같다.

bin/kafka-topics.sh -create -zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic001 (토픽생성)
bin/kafka-topics.sh --list --zookeeper localhost:2181 (토픽조회)
bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topic001 (토픽삭제)

 

producer, cunsumer 실행하자. 

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic001  (producer)
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic001 --from-beginning (cunsumer)

 

실행시 Error while fetching metadata with correlation id : {topic명}=LEADER_NOT_AVAILABLE 에러가 발생할 수 있다.

/config/server.properties에 아래와 같이 추가해주면 된다.

advertised.listeners = PLAINTEXT://localhost:9092

 

producer 콘솔창에 아무 메세지를 입력하면 cunsumer에 메세지가 수신되어 출력이 될것이다.

spring boot-kafka 구현은 다음에 이어서 하겠다.