메시지를 produce하고 consume 하는 간단한 명령어들을 알아보자.
1. Producer & Consumer 콘솔 실행
1) producer와 consumer 실행 명령어를 각각 다른 터미널 창에서 입력한다.
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first_topic -- producer 실행
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first_topic -- consumer2) producer 콘솔 창에서 메세지를 입력한다.
3) 아래와 같이 consumer 콘솔에 producer가 보낸 메세지가 출력된다.
2. 메시지를 처음부터 Consume하기
위에서 설명한 명령어로 consumer 콘솔을 실행하면 실행한 이후에 전달된 메시지만을 소비한다.
만약에 메세지를 맨 처음부터 읽고 싶다면 --from-beginning 옵션을 사용하면 된다. 아래와 같이 --from-beginning를 추가하여 명령어를 입력하면 컨슈머 콘솔을 실행하기 전에 잘못 보낸(...) 메세지까지 읽을 수 있다.
3. 메시지를 Key와 함께 Produce하기
--property parse.key=true --property key.separator 옵션을 이용해 메세지를 보낼 때 키를 설정하여 전송할 수도 있다. --property key.separator= 다음에 오는 값은 키와 값(메시지)을 구분하는 구분자가 된다.
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first_topic --property parse.key=true --property key.separator=:예시와 같이 구분자를 :로 설정하고 메시지를 보내보자.
그러면 consumer에서는 값만 전달받고 있는 것을 확인할 수 있다.
그렇다면 메시지를 전송할 때 Key는 어떤 역할을 하기에 필요한 것일까?
1) 파티셔닝 & 논리적 그룹
Kafka는 키를 해시하여 파티션을 결정한다. 이를 통해 같은 키를 가진 메시지는 항상 같은 파티션으로 전송되므로 메시지 키를 사용하면 특정 파티션에 메시지를 보낼 수 있다. 그리고 이를 이용하여 논리적인 그룹으로 메시지를 묶을 수도 있다. 예를 들어서 user의 ID를 키 값으로 사용하면 같은 user의 메시지는 항상 같은 파티션으로 전송되는 것이다.2) 키값에 따른 순서 보장
동일한 키를 가진 메시지는 동일한 파티션에 저장되기 때문에, 이 파티션에서의 메시지 순서가 보장된다. 키별로 순서를 보장하고자 할 때 유용하다.
4. 메시지를 acks 설정과 함께 Produce하기
메시지를 전달할 때 --producer-property acks= 옵션을 줌으로써 acks 모드를 설정할 수 있다. 설정 값으로는 acks=0, acks=1, acks=all(acks=-1)이 있다.
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first_topic --producer-property acks=all -- acks 설정acks란?
메시지를 브로커에 전송할 때의 확인(acknowledgement) 모드
- acks=0: 프로듀서는 브로커로부터 어떤 응답도 기다리지 않는다. 메시지가 성공적으로 전달되었는지 알 수 없기 때문에 데이터 손실 가능성이 있다.
-> 속도는 가장 빠르지만 신뢰성이 낮음.- acks=1: 리더 브로커가 메시지를 처리한 후 Producer에게 응답한다. 어느 정도의 신뢰성은 보장되지만 리더 브로커에 장애가 있을 경우 데이터 손실 가능성이 존재한다.
- acks=all (또는 acks=-1): 리더 브로커 및 리더와 동기화된 replica가 모두 메시지를 처리한 후 프로듀서에게 응답한다. 데이터 손실의 가능성이 거의 없고 가장 신뢰성이 높지만, 속도가 느려질 수 있다.
5. 기타 Producer & Consumer CLI 명령어
1) Producer CLI 명령어
- kafka-console-producer.sh --bootstrap-server localhost:9092 --topic non_existing_topic:
존재하지 않는 토픽에 메시지를 보낼 수 있다. (여기서 non_existing_topic의 자리에는 지금은 존재하지 않지만 produce와 동시에 생성하고자 하는 토픽 이름을 입력) 이렇게 토픽을 생성한 경우에는 해당 토픽의 파티션이 server.properties 파일의 num.partitions 값만큼 생성된다.
2) Consumer CLI 명령어
display key, values and timestamp in consumer
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first_topic --formatter kafka.tools.DefaultMessageFormatter --property print.timestamp=true --property print.key=true --property print.value=true --property print.partition=true --from-beginning :
메시지 format 옵션을 줄 수 있다. 위와 같은 옵션을 주는 경우에는 메시지(value)와 key, partition, timestamp까지 함께 보여준다.
