Kafka와 Spring Boot로 실시간 데이터 스트리밍 쉽게 시작하기: Pub/Sub 간단 구현

Kafka란 무엇인가?

Kafka는 대규모 실시간 데이터 스트리밍을 처리할 수 있는 분산 메시지 브로커 시스템입니다. 원래 LinkedIn에서 개발되었으며, 현재는 Apache Software Foundation에서 관리하는 오픈 소스 프로젝트입니다. Kafka는 Pub/Sub(발행/구독) 패턴을 기반으로 작동하며, 데이터를 **Producer(생산자)**가 보내고 **Consumer(소비자)**가 이를 읽습니다. 높은 처리 성능과 확장성 덕분에, Kafka는 실시간 로그 분석, 데이터 파이프라인, 스트리밍 플랫폼 등에 자주 사용됩니다.

Kafka의 주요 특징

1. 고성능: 초당 수백만 개의 메시지를 처리할 수 있는 높은 처리량 제공.
2. 내구성: 데이터를 디스크에 안전하게 저장하여, 서버 장애 시에도 데이터 손실이 없도록 보장.
3. 확장성: 클러스터 형태로 노드를 추가해 성능과 용량을 쉽게 확장할 수 있음.
4. 실시간 처리: 데이터가 실시간으로 처리되어 다양한 애플리케이션에서 즉각적으로 사용할 수 있음.
5. 다양한 언어 지원: 자바, 파이썬, Go 등 다양한 언어에서 Kafka 클라이언트를 사용할 수 있음.

반응형

Spring Boot와 Kafka 연동하여 Pub/Sub 구현하기

Spring Boot와 Kafka를 연동하여 간단한 Pub/Sub(발행/구독) 시스템을 구현해보겠습니다. 이 예제에서는 Producer가 메시지를 발행하고 Consumer가 이를 구독하여 처리하는 기본적인 구성을 다룹니다.

프로젝트 설정

1. Gradle 프로젝트 생성

Spring Initializr에서 Gradle 기반으로 새로운 프로젝트를 생성합니다. 아래와 같이 프로젝트를 설정하세요.

• Project: Gradle Project
• Language: Java
• Spring Boot: 3.x.x
• Dependencies: Spring for Apache Kafka, Spring Web, Lombok

2. Gradle 설정 (build.gradle)

Spring Boot와 Kafka 의존성을 추가하기 위해 build.gradle 파일에 다음 내용을 포함합니다.

plugins {
    id 'org.springframework.boot' version '3.1.0'
    id 'io.spring.dependency-management' version '1.1.0'
    id 'java'
}

group = 'com.example'
version = '1.0.0'
sourceCompatibility = '17'

repositories {
    mavenCentral()
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.kafka:spring-kafka'
    implementation 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
}

test {
    useJUnitPlatform()
}

Kafka 설정

1. application.yml 설정

Kafka 클러스터에 연결하기 위한 설정을 src/main/resources/application.yml 파일에 추가합니다.

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

2. Kafka Producer 구현

다음은 Kafka로 메시지를 발행하는 Producer 클래스입니다. 이 클래스는 특정 주제로 메시지를 보내는 기능을 담당합니다.

package com.example.kafka;

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

@Service
@RequiredArgsConstructor
public class KafkaProducer {

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

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

3. Kafka Consumer 구현

다음은 Kafka에서 메시지를 읽어들이는 Consumer 클래스입니다. 지정된 주제의 메시지를 구독하여 처리합니다.

package com.example.kafka;

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

@Service
public class KafkaConsumer {

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

4. Controller 추가 (API로 메시지 전송)

Spring Boot 애플리케이션에서 REST API를 통해 메시지를 Kafka로 발행할 수 있도록 간단한 컨트롤러를 추가합니다.

package com.example.kafka;

import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/kafka")
@RequiredArgsConstructor
public class KafkaController {

    private final KafkaProducer kafkaProducer;

    @PostMapping("/publish")
    public String sendMessageToKafka(@RequestParam("message") String message) {
        kafkaProducer.sendMessage(message);
        return "Message sent to Kafka topic: " + message;
    }
}

728x90

이제 Kafka와 Spring Boot를 연동하여 간단한 Pub/Sub 시스템을 구현하는 방법을 배웠습니다. 이 프로젝트는 기본적인 예제이지만, 이를 바탕으로 다양한 응용 시스템에서 실시간 데이터 스트리밍 및 분석을 할 수 있습니다. Kafka는 확장성과 성능이 뛰어나기 때문에, 더 복잡한 아키텍처에서도 효과적으로 활용될 수 있습니다.