[nodeJS] Node.js와 RabbitMQ를 사용한 메시지 큐 시스템 구축하기

RabbitMQ는 오픈 소스 메시지 브로커 소프트웨어로, 메시지 큐를 관리하고 애플리케이션 간의 메시지 전송을 지원합니다. 다양한 프로토콜과 언어를 지원하며, 확장성과 신뢰성을 제공하여 많은 기업에서 사용됩니다.

 

[RabbitMQ의 주요 기능]

• 다양한 프로토콜 지원: RabbitMQ는 AMQP, MQTT, STOMP 등 다양한 프로토콜 지원
• 확장성: 클러스터링을 통해 시스템을 확장할 수 있으며, 고가용성 보장
• 안정성: 메시지의 지속성과 안전한 전송을 보장하여 데이터 손실 방지

1. RabbitMQ와 메시지 큐의 필요성

메시지 큐란?

메시지 큐는 데이터(메시지)를 대기열에 저장하고, 이를 다른 시스템이나 애플리케이션으로 전달하는 역할
메시지 큐는 비동기 처리를 가능하게 하며, 다음과 같은 상황에서 유용하게 사용

• 비동기 작업 처리: 작업을 대기열에 넣고, 비동기적으로 처리하여 애플리케이션의 응답성을 높임.
• 로드 밸런싱: 여러 소비자 간의 작업을 분배하여 시스템의 부하를 균등하게 분산시킴.
• 내결함성: 메시지가 큐에 저장되어 있으므로, 시스템 장애 시에도 데이터 손실을 방지함.

RabbitMQ 사용 이유

• 성능: 높은 처리량과 낮은 지연 시간을 제공하여 대규모 애플리케이션에서도 안정적으로 동작
• 호환성: 다양한 프로그래밍 언어와 플랫폼을 지원하여 유연한 통합이 가능
• 커뮤니티 및 지원: 활발한 커뮤니티와 다양한 도구를 제공하여 쉽게 학습하고 구현 가능

 

2. Node.js 프로젝트에 RabbitMQ 설정하기

프로젝트 초기화 및 amqplib 설치

먼저 Node.js 프로젝트를 초기화하고, RabbitMQ와 통신하기 위해 amqplib 패키지를 설치
( amqplib는 Node.js에서 RabbitMQ와 통신하기 위한 라이브러리)

mkdir rabbitmq-nodejs
cd rabbitmq-nodejs
npm init -y
npm install amqplib

 

RabbitMQ 서버 설치 및 실행

RabbitMQ 서버를 설치하고 실행 (RabbitMQ 공식 사이트에서 운영체제에 맞는 설치 파일을 다운로드하여 설치)
설치 후, 다음 명령어로 RabbitMQ 서버 실행

rabbitmq-server

 

RabbitMQ 웹 인터페이스 활성화

RabbitMQ는 웹 기반의 관리 인터페이스를 제공
웹 브라우저에서 http://localhost:15672에 접속하여 RabbitMQ 관리 인터페이스에 접근 가능
기본 로그인 정보는 guest / guest

rabbitmq-plugins enable rabbitmq_management

 

3. 메시지 큐 생성 및 메시지 전송

메시지 큐 생성

RabbitMQ에 연결하여 task_queue라는 이름의 큐를 생성하고, Hello, RabbitMQ!라는 메시지를 큐에 전송

• amqp.connect: RabbitMQ 서버에 연결
• channel.assertQueue: 큐가 없으면 생성. durable: true 옵션은 큐를 영구적으로 유지
• channel.sendToQueue: 큐에 메시지를 전송. persistent: true 옵션은 메시지를 영구적으로 저장

// sender.js
const amqp = require('amqplib');

async function sendMessage() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    const queue = 'task_queue';
    const msg = 'Hello, RabbitMQ!';

    await channel.assertQueue(queue, {
      durable: true
    });

    channel.sendToQueue(queue, Buffer.from(msg), {
      persistent: true
    });

    console.log(`[x] Sent '${msg}'`);

    setTimeout(() => {
      connection.close();
    }, 500);
  } catch (error) {
    console.error('Error sending message:', error);
  }
}

sendMessage();

 

4. 메시지 수신 및 처리

메시지 수신

task_queue 큐에서 메시지를 수신하고, 수신된 메시지 출력
메시지를 수신한 후 channel.ack를 호출하여 메시지가 성공적으로 처리되었음을 RabbitMQ에 알림.

• channel.consume: 큐에서 메시지를 수신
• channel.ack: 메시지가 처리되었음을 확인하고 큐에서 제거

// receiver.js
const amqp = require('amqplib');

async function receiveMessage() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    const queue = 'task_queue';

    await channel.assertQueue(queue, {
      durable: true
    });

    console.log(`[*] Waiting for messages in ${queue}. To exit press CTRL+C`);

    channel.consume(queue, (msg) => {
      if (msg !== null) {
        console.log(`[x] Received '${msg.content.toString()}'`);
        channel.ack(msg); // 메시지 확인
      }
    }, {
      noAck: false // 메시지 확인을 활성화합니다.
    });
  } catch (error) {
    console.error('Error receiving message:', error);
  }
}

receiveMessage();

 

 

메시지 큐를 통해 비동기 처리를 구현하고, 시스템의 안정성과 확장성을 높일 수 있습니다. RabbitMQ는 다양한 기능과 높은 성능을 제공하여 대규모 애플리케이션에서도 효율적으로 동작합니다. 이를 통해 비동기 작업을 처리하고 시스템의 부하를 효과적으로 관리할 수 있습니다.