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

# 로컬 학습용 단일 KRaft broker (docker-compose.yml)
services:
  kafka:
    image: apache/kafka:4.3.1
    ports:
      - "9092:9092"

# 실행: docker compose up -d

npm install @nestjs/microservices @nestjs/platform-express kafkajs
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, // 토픽이 없으면 자동으로 생성
          },
          producerOnlyMode: true, // 이벤트 발행 전용 클라이언트
        },
      },
    ]),
  ],
  controllers: [UsersController],
  providers: [UsersService],
})
export class UsersModule {}

import { Body, Controller, Get, Inject, OnModuleInit, Param, Post } from '@nestjs/common';
import { ClientKafkaProxy } from '@nestjs/microservices';
import { UsersService } from './users.service';
import { lastValueFrom } from 'rxjs';

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: ClientKafkaProxy,
  ) {}

  async onModuleInit() {
    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)}`);

    // emit()은 reply topic을 만들지 않는 이벤트 발행입니다.
    // 같은 사용자의 이벤트가 같은 파티션으로 가도록 key를 함께 보냅니다.
    await lastValueFrom(
      this.client.emit('user_created', {
        key: String(newUser.id),
        value: 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 kafkajs

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
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 {}