RabbitMQ, Kafka 연동 (이벤트 기반 통신)

TCP 기반 통신은 요청-응답(Request-Response) 패턴에 적합하지만, 복잡한 분산 환경에서는 비동기적이며 이벤트 기반(Event-Based)인 통신 방식이 필요합니다. NestJS는 RabbitMQ나 Kafka와 같은 메시지 브로커를 활용하여 이를 쉽게 구현할 수 있습니다.

1. 이벤트 기반 통신의 장점

2. NestJS RabbitMQ 연동 설정RabbitMQ는 AMQP(Advanced Message Queuing Protocol) 기반의 인기 있는 메시지 브로커입니다.A. RabbitMQ 서버 (Provider) 설정RabbitMQ 연결 정보를 설정하고, 메시지를 수신하는 서버 역할을 합니다.

// src/main.ts
import { NestFactory } from '@nestjs/core';
import { Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';
import { Logger } from '@nestjs/common';

async function bootstrap() {
  const logger = new Logger('RabbitMQ Microservice');
  
  const app = await NestFactory.createMicroservice(AppModule, {
    transport: Transport.RMQ, // RabbitMQ 트랜스포트 명시
    options: {
      urls: ['amqp://user:password@localhost:5672'], // RabbitMQ 접속 URL
      queue: 'user_notifications', // 이 서비스가 메시지를 받을 Queue 이름
      queueOptions: {
        durable: false,
      },
    },
  });

  await app.listen();
  logger.log('RabbitMQ Microservice is listening...');
}
bootstrap();

B. 메시지 핸들러 구현 (컨슈머)@EventPattern() 데코레이터를 사용하여 특정 이벤트를 구독하고 처리합니다. 이벤트 기반 통신에서는 @MessagePattern() 대신 @EventPattern()을 사용합니다.

// src/notification/notification.controller.ts
import { Controller } from '@nestjs/common';
import { EventPattern, Payload } from '@nestjs/microservices';

@Controller()
export class NotificationController {

  // 'user_created_event' 이벤트가 발생하면 이 메서드가 실행됨
  @EventPattern('user_created_event') 
  handleUserCreated(@Payload() data: { userId: number, email: string }) {
    console.log(`새 사용자 이벤트 수신: ${data.userId} - ${data.email}`);
    // 실제로는 이메일 전송, 초기 알림 설정 등의 비동기 작업을 수행합니다.
  }
}

C. 클라이언트 (Publisher) 설정 및 이벤트 발행다른 서비스에서 이벤트를 RabbitMQ로 발행합니다. ClientProxy를 주입받아 .emit() 메서드를 사용합니다.

// src/user/user.service.ts
import { Injectable, Inject } from '@nestjs/common';
import { ClientProxy, Transport } from '@nestjs/microservices';

// ... 모듈 설정 시 ClientProxy 등록 필요
// ClientsModule.register([{
//   name: 'RMQ_CLIENT',
//   transport: Transport.RMQ,
//   options: { urls: ['amqp://user:password@localhost:5672'] }
// }])

@Injectable()
export class UserService {
  constructor(
    @Inject('RMQ_CLIENT') private readonly rmqClient: ClientProxy,
  ) {}

  async createUser(data: any) {
    // 1. 사용자 생성 로직 (DB 저장 등)
    // ...
    
    // 2. RabbitMQ에 이벤트 발행 (Emit)
    // .emit()은 응답을 기다리지 않습니다.
    this.rmqClient.emit('user_created_event', { userId: 123, email: data.email }); 
  }
}

3. Kafka 연동 (설정만)Kafka 연동은 RabbitMQ와 거의 동일한 패턴을 따르며, Transport.RMQ 대신 Transport.KAFKA를 사용하고 옵션을 Kafka 환경에 맞게 변경합니다.

// Kafka Transport 설정 예시
const kafkaApp = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'], // Kafka 브로커 주소
    },
    consumer: {
      groupId: 'notification-consumer-group', // 컨슈머 그룹 ID
    },
  },
});

 

특징	설명
비동기성	클라이언트가 응답을 기다리지 않고 메시지를 발행 (Publish)만 하고, 서버(Consumer)는 나중에 이를 처리합니다.
디커플링 (Decoupling)	서비스 간의 종속성이 낮아져, 한 서비스의 장애가 다른 서비스에 영향을 주지 않습니다.
확장성	메시지 브로커를 통해 여러 컨슈머 인스턴스가 메시지를 병렬 처리할 수 있어 수평 확장이 용이합니다.