[Kafka] Kafka 이용하여 Spring Boot로 메시지 보내고 받기

Kafka와 Spring Boot로 메시지 보내고 받기

안녕하세요! 오늘은 분산 메시징 시스템인 Apache Kafka를 Spring Boot와 함께 사용해 메시지를 보내고 받는 방법을 알아보려고 합니다. Kafka는 대규모 데이터 처리나 실시간 스트리밍에 강력한 도구인데요, Spring Boot와 통합하면 설정도 간단하고 코드도 깔끔해집니다. 바로 시작해볼까요?

Kafka란?

Kafka는 LinkedIn에서 개발된 오픈소스 메시징 시스템으로, 높은 처리량과 확장성을 자랑합니다. **프로듀서(Producer)**가 메시지를 보내고, **컨슈머(Consumer)**가 이를 받아 처리하는 방식으로 동작하죠. 메시지는 **토픽(Topic)**이라는 카테고리에 저장되며, 필요에 따라 여러 컨슈머가 동일한 토픽을 구독할 수 있습니다.

이제 Spring Boot에서 Kafka를 활용하는 방법을 단계별로 살펴보겠습니다.

Spring Boot에서 Kafka 사용하기

Kafka를 사용하려면 먼저 Kafka 브로커가 실행 중이어야 합니다. 로컬에서 테스트하려면 Kafka 공식 사이트에서 다운로드해 실행하거나, Docker를 사용하는 것도 추천드려요. 여기서는 기본 설정을 기준으로 진행하겠습니다.

<카프카 실행 실습>

1. 의존성 추가

pom.xml에 Spring Kafka 의존성을 추가합니다.

 

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

 

Spring Boot 버전에 따라 자동으로 적합한 Kafka 클라이언트가 포함되니, 별도로 버전을 명시하지 않아도 됩니다.

2. Kafka 설정 (application.yml)

Kafka 브로커와 연결 설정을 application.yml에 추가합니다.

spring:
  kafka:
    bootstrap-servers: localhost:9092  # Kafka 브로커 주소
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: my-group  # 컨슈머 그룹 ID
      auto-offset-reset: earliest  # 처음부터 메시지 읽기
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

• bootstrap-servers: Kafka 브로커가 실행 중인 주소.
• serializers/deserializers: 메시지의 키와 값을 문자열로 직렬화/역직렬화.

3. 메시지 보내기 (Producer)

Spring Kafka는 KafkaTemplate을 제공해 메시지 전송을 쉽게 처리할 수 있습니다. 토픽에 메시지를 보내는 서비스를 만들어보죠.

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    private static final String TOPIC = "my-topic";
    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(String message) {
        kafkaTemplate.send(TOPIC, message);
        System.out.println("Message sent: " + message);
    }
}

• KafkaTemplate: Spring이 자동으로 주입해주는 템플릿.
• send: 토픽 이름과 메시지를 인자로 전달해 전송.

4. 컨트롤러에서 Producer 호출

간단한 REST 엔드포인트를 만들어 메시지를 보내보겠습니다.

import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class KafkaController {

    private final KafkaProducerService producerService;

    public KafkaController(KafkaProducerService producerService) {
        this.producerService = producerService;
    }

    @PostMapping("/send")
    public String sendMessage(@RequestParam String message) {
        producerService.sendMessage(message);
        return "Message sent to Kafka: " + message;
    }
}

 

5. 메시지 받기 (Consumer)

이제 토픽에서 메시지를 받아오는 컨슈머를 작성합니다. @KafkaListener 어노테이션을 사용하면 매우 간단하죠.

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    @KafkaListener(topics = "my-topic", groupId = "my-group")
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }
}

• @KafkaListener: 지정한 토픽의 메시지를 감지해 메서드를 실행.
• groupId: 컨슈머 그룹을 식별. 같은 그룹 내 컨슈머는 메시지를 나누어 처리.

6. 실행 및 테스트

1. Kafka 브로커를 실행합니다 (예: bin/kafka-server-start.sh config/server.properties).
2. Spring Boot 애플리케이션을 실행합니다.
3. Postman이나 curl로 POST http://localhost:8080/send?message=Hello Kafka! 요청을 보냅니다.

그러면 콘솔에서:

• Producer: "Message sent: Hello Kafka!"
• Consumer: "Received message: Hello Kafka!"

이렇게 출력되는 걸 확인할 수 있습니다!

추가 팁: JSON 메시지 처리

문자열 대신 JSON 객체를 주고받고 싶다면, JsonSerializer와 JsonDeserializer를 사용하세요. 의존성에 jackson-databind를 추가하고 설정을 변경하면 됩니다.

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

 

application.yml 수정:

spring:
  kafka:
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"  # 역직렬화 시 신뢰할 패키지

 

그리고 DTO 클래스를 만들어 사용하면 됩니다 (예: Message 클래스).

마무리

Spring Boot와 Kafka를 함께 사용하면 메시지 기반 시스템을 정말 쉽게 구축할 수 있습니다. 이번 예제는 기본적인 Producer와 Consumer 설정을 다뤘는데요, 토픽 파티션 관리나 에러 처리 등 더 깊은 주제도 탐구해보시면 좋을 거예요. Kafka의 세계에 입문해보고 싶었다면, 지금이 시작할 완벽한 타이밍입니다!