Spring Boot를 활용한 카프카 애플리케이션 개발

이번 포스트에서는 Spring Boot를 이용해 간단하게 Kafka producer와 Kafka Consumer를 이용할 수 있는 kafka application을 개발해 보겠습니다.

 

기본 개발 환경

 

1. 의존성 & 패키지 추가

Kafka Producer & Consumer를 구현하기 위해서는 아파치 카프카 라이브러리를 프로젝트에 추가해야 합니다.

아파치 카프카 라이브러리는 mvnrepository 접속하시면 spring-kafka로 바로 접속 가능하십니다.

저 같은 경우는 2.8.0 버전의 spring-kafka 라이브러리를 이용하겠습니다.

 

만들 클래스는 Controller, Producer, Consumer 이 3가지이기 때문에 분리하기 위한 패키지 구조는 아래와 같이 잡아보았습니다.

 

2. yml 파일 설정

먼저 Kakfa Producer와 Consume에 필요한 설정 yml 파일 먼저 추가해보겠습니다. (bootstrap.yml)

spring:
    kafka:
        producer:
            bootsrap-servers: localhost:9092
            key-serializer: org.apache.kafka.common.serialization.StringSerializer
            value-serializer: org.apache.kafka.common.serialization.StringSerializer
       consumer:
           bootsrap-servers: localhost:9092
           group-id: foo
           key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
           value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

spring.kafka.producer

 - producer.bootstrap-servers : 부트스트랩 서버 설정을 localhost 카프카를 바라보도록 설정 (Broker 주소목록은 2개 이상의 IP와 Port 설정을 권장)

 - key-serializer : key값에 대한 직렬화 설정

 - value-serializer : value값에 대한 직렬화 설정

spring.kafka.consumer

 - producer.bootstrap-servers : producer 설정과 동일한 내용

 - group-id : 일명 컨슈머 그룹. 컨슈머들의 묶음

 - key-deserializer : key값에 대한 역직렬화 설정

 - value-deserializer : value값에 대한 역직렬화 설정

 

3. Kafka Controller 구현

Rest API 방식을 이용해 Kafka Message를 발행할 수 있도록 Controller.kt를 구현합니다.

@RestController
class KafkaController(
    private val kafkaProducer: KafkaProducer,
) {
    @PostMapping
    fun createMessage(
        @RequestParam("valueMessage") valueMessage: String
    ): String {
        kafkaProducer.sendTopicMessage(valueMessage)
        return "Success Create Message"
    }
}

 

4. Kafka Producer 구현

@Service
class KafkaProducer(
    val kafkaTemplate: KafkaTemplate<String, String>
) {
    private val log = LoggerFactory.getLogger(javaClass)

    fun sendTopicMessage(valueMessage: String) {
        try {
            log.info("======== Start Producer ========")
            log.info("sendMessage : $valueMessage")
            kafkaTemplate.send("create-spring-topic", valueMessage)
            log.info("======== End Producer ========")
        } catch (e: Exception) {
            e.printStackTrace()
        }
    }
}

kafkaTemplate에 토픽명과 메시지를 전달할 수 있습니다.

직접 로그도 찍어보면서 메시지가 잘 전달이 되는지 확인해봅니다.

 

5. Kafka Consumer 구현

@Service
class KafkaConsumer {
    private val log = LoggerFactory.getLogger(javaClass)

    @KafkaListener(topics = ["create-spring-topic"], groupId = "foo")
    fun consumeMessage(message: String) {
        log.info("======== Start Consumer ========")
        log.info("Receive Consumer Message : $message")
        log.info("======== End Consumer ========")
    }
}

여기서 @KafkaListener라는 어노테이션을 이용했는데 해당 어노테이션은 특정 Topic에 대한 Kafka message의 Consumer가 될 메소드를 지정해줍니다. 내부에 어떤 것들이 구현되어있는지 확인해보는 것도 추천드립니다.

 

6. Test

이제 실제 Postman을 이용해 해당 API를 사용해보도록 해보겠습니다.

Kafka Producer에서 기본 경로와 valueMessage라는 쿼리스트링을 지정해줘서 아래와 같이 POST 요청을 하면 200 OK와 같이 메시지가 성공적으로 보내졌다는 메시지를 볼 수 있을 것입니다.

그리고 서버 로그도 제대로 뜨는 것도 있습니다.

 

그리고 docker로 띄운 kafka 컨테이너에 접속해 실제로 Topic도 잘 발행되었는지 확인해봅니다.

# kafka-topics.sh --bootstrap-server localhost:9092 --list

실제 Topic도 잘 생성되었습니다. 이렇게 Topic이 자동적으로 생성될 수 있는 이유는 Broker 설정 중에 자동으로 Topic을 생성해주는 auto.create.topics.enable 옵션이 있는데 default 값이 true이기 때문에 자동으로 생성이 가능합니다.

 

 

참고 문헌 😶

https://docs.spring.io/spring-kafka/reference/html/#preface

Spring for Apache Kafka

https://kafka.apache.org/documentation/#gettingStarted

Apache Kafka

https://docs.spring.io/spring-kafka/api/org/springframework/kafka/annotation/KafkaListener.html