📌 NestJS + Kubernetes & 마이크로서비스 배포: 실전 가이드 - NestJS + Kafka 실시간 데이터 처리

📌 NestJS + Kubernetes & 마이크로서비스 배포: 실전 가이드

25. NestJS + Kafka 실시간 데이터 처리

---

데이터를 실시간으로 수집하고 처리해야 하는 시스템에서는 Kafka와 같은 스트리밍 플랫폼이 필수입니다.
Kafka는 대용량 데이터를 고속으로 처리하며, 데이터 파이프라인 구축에 적합합니다.
이번 글에서는 NestJS와 Kafka를 결합하여 실시간 데이터 스트리밍을 처리하는 구조를 설계하고,
Kafka와 NestJS를 Kubernetes 환경에서 효과적으로 운영하는 방법을 소개합니다. 🌐

---

✅ 1. Kafka 기반 실시간 데이터 파이프라인 구조

[Producer] → [Kafka Topic] → [Consumer] → [NestJS Service] → [Database/Storage]

💡 Kafka의 역할

• 대용량 로그 및 이벤트 처리
• 마이크로서비스 간 데이터 스트리밍
• 비동기 데이터 파이프라인 구성

---

✅ 2. Kafka 설치 및 설정

🔸 로컬 Docker Kafka 설치 (Bitnami 이미지)

docker run -d --name kafka-server \
  -p 9092:9092 -p 2181:2181 \
  -e ALLOW_PLAINTEXT_LISTENER=yes \
  bitnami/kafka:latest

💡 Kafka UI 관리 도구:

docker run -d -p 9000:9000 \
  -e KAFKA_BROKERS=kafka-server:9092 \
  provectuslabs/kafka-ui:latest

---

✅ 3. NestJS + Kafka 기본 설정

🔸 필수 패키지 설치

npm install @nestjs/microservices kafkajs

---

📂 Kafka 모듈 구성

📂 kafka/kafka.module.ts

import { Module } from '@nestjs/common';
import { KafkaService } from './kafka.service';

@Module({
  providers: [KafkaService],
  exports: [KafkaService],
})
export class KafkaModule {}

---

📂 Kafka 서비스 구현

반응형

📂 kafka/kafka.service.ts

import { Injectable, OnModuleInit } from '@nestjs/common';
import { Kafka, Consumer, Producer } from 'kafkajs';

@Injectable()
export class KafkaService implements OnModuleInit {
  private producer: Producer;
  private consumer: Consumer;

  async onModuleInit() {
    const kafka = new Kafka({
      clientId: 'nestjs-kafka-client',
      brokers: ['localhost:9092'],
    });

    this.producer = kafka.producer();
    this.consumer = kafka.consumer({ groupId: 'nestjs-group' });

    await this.producer.connect();
    await this.consumer.connect();

    await this.consumer.subscribe({ topic: 'nestjs-topic', fromBeginning: true });
    await this.consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        console.log(`Received message: ${message.value.toString()}`);
      },
    });
  }

  async sendMessage(topic: string, message: string) {
    await this.producer.send({
      topic,
      messages: [{ value: message }],
    });
  }
}

---

✅ 4. Kafka 메시지 전송 및 수신

📂 API 호출 예시 (컨트롤러)

📂 app.controller.ts

@Controller('kafka')
export class AppController {
  constructor(private readonly kafkaService: KafkaService) {}

  @Post('send')
  async sendMessage(@Body('message') message: string) {
    await this.kafkaService.sendMessage('nestjs-topic', message);
    return { status: 'Message sent' };
  }
}

💡 요청 예시

curl -X POST http://localhost:3000/kafka/send -d '{"message": "Hello Kafka"}' -H "Content-Type: application/json"

---

✅ 5. Kafka와 Kubernetes 연동

📂 Helm으로 Kafka 클러스터 배포

helm repo add bitnami https://charts.bitnami.com/bitnami
helm install kafka bitnami/kafka

📂 Kafka 서비스 확인

kubectl get pods
kubectl get svc

---

📂 Kubernetes용 Kafka 설정

📂 config/kafka-configmap.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-config
data:
  KAFKA_BROKER: kafka:9092

📂 deployment.yaml

env:
  - name: KAFKA_BROKER
    valueFrom:
      configMapKeyRef:
        name: kafka-config
        key: KAFKA_BROKER

---

✅ 6. 실시간 데이터 처리 예시

데이터 유형 Kafka 토픽 처리 방법

사용자 로그인	user-login	실시간 로그 분석 및 모니터링
주문 생성	order-created	주문 처리 및 결제 상태 업데이트
AI 분석 결과	ai-results	모델 예측 결과를 DB에 저장 및 대시보드 표시

---

📂 실시간 처리 로직 (NestJS)

📂 order-consumer.service.ts

@ConsumerService('order-created')
export class OrderConsumerService {
  async handleMessage(message: string) {
    console.log(`Order Received: ${message}`);
    // 주문 처리 로직
  }
}

---

✅ 7. Kafka 모니터링 및 운영 전략

항목 도구

브로커 상태 확인	Kafka UI, Kafka Manager
메시지 모니터링	Kafka-UI, Prometheus + Grafana
장애 대응	Kafka MirrorMaker, Multi-Broker 구성
로그 수집	ELK 스택 (Elasticsearch, Logstash, Kibana)

---

✅ 8. 실시간 데이터 파이프라인 활용 사례

1. 이벤트 스트리밍: 실시간 유저 행동 로그 수집
2. 데이터 파이프라인: 로그 수집 → Kafka → 데이터 분석 시스템
3. 알림 서비스: 특정 이벤트 발생 시 실시간 푸시 알림
4. AI 실시간 추론 결과: Kafka로 처리 결과 전달

---

✅ 결론: Kafka 기반 NestJS 실시간 데이터 처리 완성

✅ Kafka를 통해 대용량 실시간 데이터를 안정적으로 처리
✅ NestJS와 연계하여 이벤트 기반 아키텍처 구현
✅ Kubernetes에 Kafka를 배포하여 확장성 확보
✅ Kafka 클러스터와 NestJS 간의 서비스 간 통신 구조 최적화

다음 글에서는 NestJS를 기반으로 GraphQL API를 고도화하여 성능을 극대화하는 전략을 소개합니다! 💡

---

🔍 다음 글 예고: NestJS GraphQL 성능 최적화 – 데이터 로딩과 캐싱 전략

📌 다음 편: 26. NestJS GraphQL 고성능 API 구축

---

 

NestJS Kafka,NestJS 실시간 데이터,NestJS 스트리밍,NestJS Kafka 클러스터,NestJS Kafka 설정,NestJS Kafka Kubernetes,NestJS Kafka Helm,NestJS Kafka Consumer,NestJS Kafka Producer,NestJS 실시간 메시지 처리