Kafka 메시지 생산/소비 예제

Topic

• Topic 관련 옵션이 궁금할 때: bin/kafka-topics.sh --help (help 명령어)

• Topic 생성
  bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

  • --create: 새로운 토픽 생성 옵션
  • --bootstrap-server:
    • 연결할 Kafka 서버(host:port)
      • 해당 옵션으로 인해, Zookeeper에 직접 연결할 필요가 없다.
  • --replication-factor:
    • Partition 복제 수
      • 해당 옵션을 설정하지 않으면 기본값 사용
        - 기본값: server.properties의 default.replication.factor로 설정 가능(/opt/kafka/config 경로에 server.properties 존재)
  • --partitions
    • Topic이 생성되거나 변경될 때 Partition 수
      • 해당 옵션을 설정하지 않으면 기본값 사용
        - 기본값: server.properties의 num.partitions로 설정 가능
  • --topic
    • create, alter, describe, delete 옵션에 사용할 토픽 이름
      • Topic명은 큰따옴표("")로 묶고 정규표현식이 가능하므로 \로 escape.

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

• Topic 상세정보
  bin/kafka-topics.sh --describe --topic {topic명} --bootstrap-server localhost:9092

• Topic 삭제
  bin/kafka-topics.sh --delete --topic {topic명} --bootstrap-server localhost:9092
  토픽을 삭제하려면 server.properties의 delete.topic.enable=true 설정으로 가능 + 서버 재시작

Producer

• Producer 관련 옵션이 궁금할 때: bin/kafka-console-producer.sh --help (help 명령어)
• 메세지 생산
  bin/kafka-console-producer.sh --broker-list localhost:9092 --topic {topic명}

Consumer

• Consumer 관련 옵션이 궁금할 때: bin/kafka-console-consumer.sh --help (help 명령어)

• 메세지 소비
  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic {topic명} --from-beginning

• --from-beginning
  - Consumer에서 설정된 offset이 없으므로 가장 최신의 메시지 대신 가장 먼저 도착한 메시지부터 읽도록 하는 옵션