[Spring Boot] Kafka 연동 간단 예제

Kafka를 연동하여 아주 간단한 pub/sub 구조의 서비스를 만들어본다.

스프링 서버로 message 요청이 오면 이를 Kafka 서버로 전달하고, Kafka는 해당 토픽을 구독중인 컨슈머에게 전달하는 아주 간단한 구조이다.

 

# Kafka 설치

윈도우 11기준으로 2.8.0 버전을 설치해서 사용했다.

적당한 위치에 설치하고 폴더명을 kafka (혹은 짧은) 명으로 바꿔준다. 

나중에 cli로 명령어를 입력할 때 너무 명령어가 길다는 오류를 보지 않기 위함이다.

 

# Zookeeper 와 Kafka 실행

Kafka는 Zookeeper를 내부적으로 사용한다. Kafka를 실행하기 위해선 반드시 먼저 Zookeeper를 켜줘야 한다.

Kafka 설치 경로로 이동하여 cmd를 열어준다.

Zookeeper를 실행한다. cmd 창에 다음 명령어를 입력한다.

bin\windows\zookeeper-server-start.bat config\zookeeper.properties

현재 cmd창을 열어두고 새로운 창을 켜서 Kafka 서버를 띄워준다. 

bin\windows\kafka-server-start.bat config\server.properties

또 cmd 창을 열어서 netstat -a 를 입력한 후 정상적으로 가동중인지 확인한다.

Zookeeper 기본 포트는 2181, Kafka는 9092다.

해당 명령어를 입력하면 더 쉽게 찾을 수 있다.

# Spring Boot 프로젝트 세팅

• Spring Web
• Spring for Apache Kafka
• Kotlin
• Gradle (Kotlin)

application.yml 파일을 생성해 준다.

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
      group-id: group-id-mopil # 컨슈머를 유일하게 식별할 수 있는 값
      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

디렉토리 구조

 

ProducerConsumer.kt

@Component
class SimpleConsumer {
    @KafkaListener(topics = ["mopil"], groupId = "group-id-mopil")
    fun consume(message: String) {
        println("receive message : $message")
    }
}

data class MessageDto(val message: String)

@RestController
class SimpleProducer(
    private val kafkaTemplate: KafkaTemplate<String, String>
) {

    @PostMapping("/message")
    fun sendMessage(@RequestBody messageDto: MessageDto) {
        println("send message : $messageDto.message")
        kafkaTemplate.send("mopil", messageDto.message)
    }
}

 

이제 스프링 부트 서버를 작동시키고 포스트맨으로 요청을 보내면 Kafka 서버로 이동되고, 컨슈머는 이를 감지해서 소비한다.

 

동일한 스프링 부트 서버이지만, 분명 Kafka를 통해 메시지를 전달받았다.

https://github.com/mopil/spring-boot-kafka-simple

GitHub - mopil/spring-boot-kafka-simple: Spring boot + Kafka 간단 예제