[Kafka] Apache Kafka에 대해서 알아보자

Kafka를 알기 전에는 MQ(Message Queue)를 알아야 한다..!

메시지 큐(MQ)

• 비동기 메시지를 사용하는 다른 응용프로그램 사이의 데이터 송수신을 의미하는데 MOM을 구현한 시스템 => MQ(메세지 큐)
• * 메시지 지향 미들웨어(Message Oriented Middleware: MOM)
• Producer가 메시지를 큐에 전송하면 Consumer가 처리하는 방식
• 메시지를 비동기로 처리하고 queue에 저장하기 때문에 consumer에게 병목을 줄여줄 수 있는 장점이 있음

---

Kafka 란?

• 분산형 스트리밍 플랫폼
• 메시지를 메모리에 저장하는 메시징 시스템과는 달리 파일에 저장을 하기 때문에 카프카를 재시작해도 메시지 유실 우려가 감소된다.
• 브로커가 컨슈머에게 메시지를 push해주는 방식이 기본 메시징 시스템이라면, 카프카는 컨슈머가 브로커로부터 메시지를 직접 가져가는 pull방식으로 동작 (따라서 컨슈머는 자신의 처리 능력만큼의 메시지만 가져와 최적의 성능을 낼 수 있음)

• 하드디스크의 순차적 읽기에 대한 성능은 메모리보다 크게 떨어지지 않는다. (카프카의 파일 시스템을 활용한 고성능 디자인에 근거)

---

kafka라는게 어떤 기술인지는 어느정도 알겠다... 그럼 왜 사용하는가?

1. 발신자와 수신자 서로가 서로에게 의존하지 않음

2. 수신자가 장애 상황이 와도 발행된 메세지가 큐에 남아있기 때문에 수신자에게 특별한 이슈가 있지 않은 이상 무조건 메세지가 전달됨

3. 비동기 통신을 구현할 수 있는 점이 가장 메리트가 있음(높은 처리량, 낮은 지연시간)

---

Kafka 실제로 사용해보기

[ kafka 명령어 ]

* 각 명령어의 경로(./bin~)는 카프카 설치 경로에 따라 다를 수 있음

나는 docker를 통해서 kafka이미지를 pull받아 사용했다. kafka를 설치한 위치에서 명령어를 입력하면 된다.

• 토픽 목록 확인

$ ./bin/kafka-topics.sh —bootstrap-server localhost:9092 --list

(* 9092는 카프카의 default port이다.)

 

• 토픽 정보 확인

$ ./bin/kafka-topics.sh --describe —topic [확인할 topic 이름] —bootstrap-server localhost:9092

 

• 토픽 생성

$ ./bin/kafka-topics.sh --create --topic <topic-name> --bootstrap-server <broker-list> [--replication-factor <replication-factor>] [--partitions <num-partitions>]

ex) ./bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --replication-factor 1

• 토픽 삭제

$ ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic 토픽이름

 

!!! topic이 delete되지 않는다면? !!!

-> server.properties에 delete.topic.enable = true 옵션을 추가해주세요..

(* server.properties는 높은 확률로 /opt/kafka/bin안에 있을 것임)

 

그래도 안된다면...

- kafka cluster의 로그폴더 내용을 삭제 후 재시작합니다.

- --delete뒤에 --force명령어를 추가 입력하면 지워진 토픽정보를 복구하지 않고 delete됩니당

 

• producer 실행 (데이터 입력)

$ ./bin/kafka-console-producer.sh --topic 토픽이름 —broker-list localhost:9092

 

또 다른 터미널 실행 후...

• consumer 실행

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic 토픽이름 --from-beginning

 

이제 메세지를 보내보세용

그러면 바로바로 메세지가 뜨게 된답니다~!

---

kafka 메시지 삭제 정책
"Kafka는 메세지를 큐에 넣는데 그럼 삭제는 못하나요..??"

• kafka는 파티션에 데이터를 저장함
• 카프카를 설정하면 파티션이 저장될 디렉토리의 위치를 log.dirs 매개변수에 지정할 수 있음
• 다른 메시징 시스템과 달리 토픽의 메시지가 소비된 후 즉시 삭제하지 않음
• 각 토픽의 구성에 따라 허용되는 공간과 관리 방식이 결정된다.

1. 특정 오프셋이나 설정된 byte가 넘어가면 새로운 segment가 생성되어 메시지가 관리된다.
2. 시간기반, 크기기반으로 오래된 세그먼트를 삭제한다.
3. 시간 기반 보유: Segments에 대해 retention time에 도달하면 정의된 정책에 따라 segments가 삭제 또는 압축 대상으로 표시된다. (기본 보존 기간은 7일)

-> retention.hours, minutes, ms 등으로 런타임에서 값을 수정할 수 있음

1. 크기 기반 보유: 토픽 파티션에 대한 로그 데이터 구조의 최대 크기를 설정할 수 있음. 로그의 크기가 설정에 도달하면 마지막부터 세그먼트 제거

-> 디스크 공간이 제한되는 상황에서는 유용(세그먼트가 하나밖에 없는 경우에는 정상적으로 삭제되지 않을 수 있음)

 

=> 나는 그냥 토픽을 삭제하고 다시 생성하는 방식으로 진행했다. (어차피 나중되면 사라질 내용..그냥 지금 삭제한다 생각하고...ㅎ)

 

 

 

Kafka를 spring에 녹여낸 내용은 다음에 포스팅하도록 하겠다,,,,