[springboot] springboot kafka 연동하기

docker compose 로 설치해 버리면 편하지만 kafka 가 어떻게 동작하는지 이해하려면 파일을 다운로드 해서 설치해보자

그 후 docker 를 이미지로 설치하는 과정으로 구현

 

MacOS 에 kafka 를 설치 

https://ldh-6019.tistory.com/228?category=1059128 

[kafka] MacOs Kafka install / test

 

명령어 모음  kafka 실행

#카프카 설치 경로로 이동 
cd /Users/doo/kafka/kafka_2.13-2.8.0
#주키퍼 시작
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
#카프카 시작
bin/kafka-server-start.sh -daemon config/server.properties
#토픽생성
bin/kafka-topics.sh --create --topic 900gle --bootstrap-server localhost:9092
#프로듀서 실행
bin/kafka-console-producer.sh --topic 900gle --bootstrap-server localhost:9092
#컨슈머 실행
bin/kafka-console-consumer.sh --topic 900gle --from-beginning --bootstrap-server localhost:9092

 

springboot

build.gradle

/* kafka */
implementation 'org.springframework.kafka:spring-kafka'

application.yml

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
      group-id: doo
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      bootstrap-servers: localhost:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

consumer와 producer 설정

• spring.kafka.consumer
  •  bootstrap-servers
    • Kafka 클러스터에 대한 초기 연결에 사용할 호스트:포트쌍의 쉼표로 구분된 목록입니다.
    • 글로벌 설정이 있어도, consumer.bootstrap-servers가 존재하면 consuemer 전용으로 오버라이딩 
  • group-id
    • Consumer는 Consumer Group이 존재하기 때문에, 유일하게 식별 가능한 Consumer Group을 작성
  • auto-offset-reset
    • Kafka 서버에 초기 offset이 없거나, 서버에 현재 offset이 더 이상 없는 경우 수행할 작업을 작성합니다.
    • Consumer Group의 Consumer는 메시지를 소비할 때 Topic내에 Partition에서 다음에 소비할 offset이 어디인지 공유를 하고 있습니다. 그런데 오류 등으로 인해. 이러한 offset 정보가 없어졌을 때 어떻게 offeset을 reset 할 것인지를 명시
      • latest : 가장 최근에 생산된 메시지로 offeset reset
      • earliest : 가장 오래된 메시지로 offeset reset
      • none : offset 정보가 없으면 Exception 발생
    • 직접 Kafka Server에 접근하여 offset을 reset할 수 있지만, Spring에서 제공해주는 방식은 위와 같음.
  • key-deserializer / value-deserializer
    • Kafka에서 데이터를 받아올 때, key / value를 역직렬화 
    • 여기서 key와 value는 뒤에서 살펴볼 KafkaTemplate의 key, value를 의미
    • 이 글에서는 메시지가 문자열 데이터이므로 StringDeserializer를 사용했습니다. JSON 데이터를 넘겨줄 것이라면 JsonDeserializer도 가능
• spring.kafka.producer
  •  bootstrap-servers
    
    • consumer.bootstrap-servers와 동일한 내용이며, producer 전용으로 오버라이딩 하려면 작성
  • key-serializer / value-serializer
    • Kafka에 데이터를 보낼 때, key / value를 직렬화 
    • consumer에서 살펴본 key-deserializer, value-deserializer와 동일