Apache Kafka와 이벤트 기반 아키텍처

# Docker Compose 파일 (docker-compose.yml) 예시
version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
      - "9093:9093" # 내부 통신용 포트
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9093,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    depends_on:
      - zookeeper

# 실행: docker-compose up -d

npm install @nestjs/microservices @nestjs/platform-express
npm install --save-dev @types/express # 웹 서버 역할을 겸할 경우

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  // NestJS 마이크로서비스 챕터의 컨벤션을 위해 포트 3000 사용
  await app.listen(3000);
  console.log('Users Service (HTTP Server) is listening on port 3000');
}
bootstrap();

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { UsersController } from './users.controller';
import { UsersService } from './users.service';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'KAFKA_SERVICE', // Kafka 클라이언트 토큰
        transport: Transport.KAFKA, // Kafka 전송 방식 사용
        options: {
          client: {
            clientId: 'user-producer', // 클라이언트 ID
            brokers: ['localhost:9092'], // Kafka 브로커 주소
          },
          producer: {
            allowAutoTopicCreation: true, // 토픽이 없으면 자동으로 생성
          },
        },
      },
    ]),
  ],
  controllers: [UsersController],
  providers: [UsersService],
})
export class UsersModule {}

import { Controller, Post, Body, Get, Param, Inject } from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices'; // ClientKafka 임포트 (ClientProxy의 하위 타입)
import { UsersService } from './users.service';
import { OnModuleInit } from '@nestjs/common'; // OnModuleInit 임포트
import { lastValueFrom } from 'rxjs'; // lastValueFrom 임포트

interface User {
  id: number;
  name: string;
  email: string;
}

@Controller('users') // HTTP 엔드포인트
export class UsersController implements OnModuleInit {
  constructor(
    private readonly usersService: UsersService,
    @Inject('KAFKA_SERVICE') private readonly client: ClientKafka, // Kafka 클라이언트 주입
  ) {}

  async onModuleInit() {
    // 이 클라이언트가 메시지를 발행할 모든 토픽을 구독해야 합니다.
    // NestJS는 이 토픽을 미리 연결하여 메시지 전송 실패를 방지합니다.
    this.client.subscribeToResponseOf('user_created');
    await this.client.connect(); // 클라이언트 연결
  }

  @Post()
  async createUser(@Body() userDto: { name: string; email: string }): Promise<User> {
    const newUser = this.usersService.create(userDto);
    console.log(`Users Service: New user created: ${JSON.stringify(newUser)}`);

    // 'user_created' 토픽으로 이벤트 발행
    // emit()은 응답을 기다리지 않는 이벤트 기반 통신에 사용합니다.
    this.client.emit('user_created', newUser); // 메시지 키는 생략하고 값만 전송

    return newUser;
  }

  // 기존 사용자 조회 엔드포인트 (HTTP)
  @Get(':id')
  getUserById(@Param('id') id: string): User {
    const userId = parseInt(id, 10);
    return this.usersService.findOne(userId);
  }
}

nest new notifications-service --skip-install
cd notifications-service
npm install @nestjs/microservices
npm install

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { join } from 'path';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
    transport: Transport.KAFKA, // Kafka 전송 방식 사용
    options: {
      client: {
        clientId: 'notification-consumer', // 클라이언트 ID
        brokers: ['localhost:9092'], // Kafka 브로커 주소
      },
      consumer: {
        groupId: 'notification-group', // 컨슈머 그룹 ID (필수)
      },
    },
  });
  await app.listen();
  console.log('Notifications Microservice (Kafka Consumer) is listening.');
}
bootstrap();

import { Controller } from '@nestjs/common';
import { EventPattern, Payload } from '@nestjs/microservices'; // EventPattern, Payload 임포트
import { NotificationsService } from './notifications.service';

interface UserCreatedEvent { // user_created 이벤트의 페이로드 구조 정의
  id: number;
  name: string;
  email: string;
}

@Controller()
export class NotificationsController {
  constructor(private readonly notificationsService: NotificationsService) {}

  // 'user_created' 토픽의 이벤트를 처리하는 핸들러
  @EventPattern('user_created')
  async handleUserCreated(@Payload() data: UserCreatedEvent) {
    console.log(`Notifications Service: Received 'user_created' event for user: ${JSON.stringify(data)}`);
    // 여기에서 실제 알림 로직 (예: 이메일 전송)을 구현합니다.
    await this.notificationsService.sendWelcomeEmail(data.email, data.name);
  }
}

import { Injectable } from '@nestjs/common';

@Injectable()
export class NotificationsService {
  async sendWelcomeEmail(email: string, name: string): Promise<void> {
    console.log(`Sending welcome email to ${name} (${email})...`);
    // 실제 이메일 전송 로직 (예: Nodemailer, SendGrid 등 사용)
    await new Promise(resolve => setTimeout(resolve, 1000)); // 1초 지연 시뮬레이션
    console.log(`Welcome email sent to ${name}!`);
  }
}

import { Module } from '@nestjs/common';
import { NotificationsController } from './notifications.controller';
import { NotificationsService } from './notifications.service';

@Module({
  controllers: [NotificationsController],
  providers: [NotificationsService],
})
export class NotificationsModule {}

import { Module } from '@nestjs/common';
import { NotificationsModule } from './notifications/notifications.module';

@Module({
  imports: [NotificationsModule],
  controllers: [],
  providers: [],
})
export class AppModule {}