🤠 RabbitMQ로 분산 서버에게 메시지를 분배하기

분야	작성자	작성일
BE	김민수	24년 11월 17일

RabbitMQ로 분산 서버에게 메시지를 분배하기

개요

현재 프로젝트에서 실시간으로 한국투자증권 API로부터 데이터를 받고 있지만, 단일 인스턴스로는 한계를 느끼게 되었고 이를 극복하기 위해 오토 스케일링을 고려하게 되었다. 이때 각 서버마다 받아야하는 주식을 할당하기 위한 방법을 고민하다가 메시지 큐를 활용하여 분배하기로 했다.

RabbitMQ 이란?

오픈 소스 메시지 브로커로, 애플리케이션 간 메시지를 송수신하는데 사용된다. 주로 메시지 큐 역할을 수행한다.

핵심 개념

• Publisher: 메시지를 보내는 주체
• Consumer: Publisher 로부터 메시지를 받아 처리하는 주체
• Exchange: Publisher 로부터 전달 받은 메시지를 큐로 전달하는 곳
• Queue: Consumer 가 메시지를 소비하기 전까지 보관하는 장소
• Binding: Exchange 와 Queue 의 관계, 보통 사용자가 특정 exchange 가 특정 queue 를 binding 하도록 정의한다.

RabbitMQ 설치 후 실행

우선 rabbitMQ를 docket에 설치하기 위해 아래의 명령어를 통해서 이미지를 다운받은 후 컨테이너를 실행했다.

// rabbitmq 이미지 설치
docker pull rabbitmq

// 컨테이너 실행
docker run -d -p 15672:15672 -p 5672:5672 --name rabbitmq rabbitmq

// rabbitmq_management 설치
docker exec rabbitmq rabbitmq-plugins enable rabbitmq_management

이후 localhost:15672 에 접속하면 관리자 페이지에 접속할 수 있다. 이때 아이디와 비밀번호는 모두 guest이다.

생산자 만들기

우선 생산자는 간단하게 node만으로 만들었다. amqplib 라이브러리를 통해서 큐를 정하여 메시지를 전달하도록 했다.

먼저 RabbitMQ 서버에 연결하고 채널을 생성한다. 이때 amqp://localhost 은 localhost의 rabbitMQ의 기본 포트(5672)로 접속하게 되므로 포트가 다르면 수정해야한다.

const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();

채널을 생성한 후 assertQueue로 큐를 할당한다. 이때 큐가 없으면 생성한다.

await channel.assertQueue(queue, {
    durable: true,  // 큐가 서버가 재시작해도 유지되도록 설정
});

이후 원하는 로직을 실행한다. 이번에 진행할 로직은 100개의 메시지를 큐에 넣는 것이다.

// 메시지 전송
for (let i = 0; i< 100; i++) {
    const message = `Hello World! ${i}`;
    channel.sendToQueue(queue, Buffer.from(Buffer.from(
        JSON.stringify({
            pattern: 'hello',
            data: {source: message},
        })
    )), {
        persistent: true
    });
    console.log(`sent: ${message}`);
}

메시지 내용은 pattern과 data로 구성되어 있다.

• pattern: 이벤트 패턴을 나타내며 nestjs에서 특정 이벤트 패턴 메시지를 받게된다.
• data: 전달하는 메시지

또한 메시지를 큐에 전달할 때 persistent라는 옵션을 설정했는데 메시지 브로커가 재시작해도 유지하는 옵션이다.

전체 코드

const amqp = require('amqplib');

async function sendTask() {
    const queue = 'kospi';

    try {
        // RabbitMQ 서버에 연결
        const connection = await amqp.connect('amqp://localhost');
        const channel = await connection.createChannel();

        // 큐 생성 (없으면 생성)
        await channel.assertQueue(queue, {
            durable: true,  // 큐가 서버가 재시작해도 유지되도록 설정
        });

        // 메시지 전송
        for (let i = 0; i< 100; i++) {
            const message = `Hello World! ${i}`;
            channel.sendToQueue(queue, Buffer.from(Buffer.from(
                JSON.stringify({
                    pattern: 'hello',
                    data: {source: message},
                })
            )), {
                persistent: true,
            });
            console.log(`sent: ${message}`);
        }

        // 연결 종료
        await channel.close();
        await connection.close();
    } catch (error) {
        console.error('Error sending task:', error);
    }
}
sendTask();

소비자 만들기

소비자는 기존 서버가 사용하는 nestjs로 구성하도록 했다. 먼저 연관된 패키지를 설치한다.

npm i --save amqplib amqp-connection-manager @nestjs/microservices

이후 main 파일에서 마이크로서비스를 설정한 후 시작하도록 한다.

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.connectMicroservice<MicroserviceOptions>({
    transport: Transport.RMQ,
    options: {
      urls: ['amqp://localhost'],
      queue: 'kospi',
      noAck: false,
      queueOptions: {
        durable: true,
      },
    },
  });
  await app.startAllMicroservices();
  await app.listen(process.env.PORT ?? 3000);
}

bootstrap();

이때 rabbitmq와 관련된 옵션을 설정했는데 각 옵션은 다음과 같다.

• urls - rabbitmq 주소
• queue - 큐 이름
• noAck - 로직이 완료된 후 ack 호출 없이 자동으로 보내는 옵션이다.
• queueOptions - 큐와 연관된 옵션

이후 생산자에 전달하는 이벤트 패턴 메시지를 받는 컨트롤러를 구현한다.

@Controller()
export class AppController {
  @EventPattern('hello')
  async handleMessage(@Payload() data: number[], @Ctx() context: RmqContext) {
    const channel = context.getChannelRef();
    const originalMsg = context.getMessage();
    console.log('data:', data);
    channel.ack(originalMsg);
  }
}

이후 2개의 nestjs 애플리케이션을 실행 후 100개의 메시지를 받을 때 아래와 같이 잘 분배가 되는 것을 볼 수 있다.

만약 ACK를 실행하지 않으면?

이전에 noAck를 false로 설정했다. 이렇게 한 이유는 로직이 실패했을 때 메시지가 사라지지 않고 다시 메시지 큐로 전달하기 위함이다.

noAck를 false로 한 상태로 ack를 호출하지 않으면 자동으로 rabbitMQ로 돌아갈 것이라 예상되지만 그렇지 않다. 우선 이전의 코드에서 0.5확률로 ack를 호출하지 않도록 하고, 1개의 서버만 실행한다

@EventPattern('hello')
async handleMessage(@Payload() data: number[], @Ctx() context: RmqContext) {
  const channel = context.getChannelRef();
  const originalMsg = context.getMessage();
  if (Math.random() > 0.5) {
    return;
  }
  console.log('data:', data);
  channel.ack(originalMsg);
}

이후 실행했을 때 아래의 결과처럼 일부 메시지가 ack 가 되지 않는 것을 확인할 수 있다.

그렇다면 RabbitMQ에 메시지가 돌아왔는지 확인해보자

RabbitMQ 에서는 모든 메시지가 전달되어 비어있는 상태이다. 이번에는 서버를 강제로 중지하고 다시 확인해보자

놀랍게도 서버가 종료될 때 다시 메시지가 채워진 것을 볼 수 있다. 해당 메시지는 아마도 ack가 호출되지 않은 메시지로 보인다. 이러한 결과를 통해서 ack 를 호출하지 않으면 서버에 계속 쌓이게 되어 메모리가 낭비되게 된다.

그렇다면 메시지에 대한 로직이 실패하여 메시지 큐로 다시 되돌릴 때 어떻게 해야하는가? 바로 nack 를 호출하면 된다. 해당 메서드의 시그니처는 다음과 같다.

channel.nack(message, allUpTo, requeue);

• message: 메시지 객체이다. 채널에 전송된 메시지를 참조한다.
• allUpTo (boolean): true로 설정하면, 주어진 메시지 까지의 모든 이전 메시지들도 함께 부정 확인된다. 기본값은 false
• requeue (boolean): true로 설정하면, 메시지를 큐의 처음으로 다시 넣어 재처리할 수 있도록 한다. false로 설정하면 메시지가 버려진다. 기본값은 true

이를 참고하여 다음과 같이 코드를 수정했다.

@EventPattern('hello')
async handleMessage(@Payload() data: number[], @Ctx() context: RmqContext) {
  const channel = context.getChannelRef();
  const originalMsg = context.getMessage();
  if (Math.random() > 0.5) {
    return channel.nack(originalMsg, false, true);
  }
  console.log('data:', data);
  channel.ack(originalMsg);
}

이후 아래의 결과를 통해 ack가 되지 않는 메시지는 큐에 다시 채워지고 재처리되는 것을 볼 수 있다.