RabbitMq Tutorial

RabbitMQ는 AMQP 기반 오픈소스 메시지 브로커이다.
메시지 브로커란 메시지 큐들을 관리해 서로 다른 시스템, 컴포넌트간에 효율적인 메시지를 교환을 도와준다.
이를 통해 시스템에 높은 신뢰성을 제공하고, 느슨한 연결을 통해 시스템의 손쉬운 확장을 가능케 한다.

Message Queue

메시지 큐는 단순히 설명하면 선입 선출을 통해 데이터의 순서를 보장하는 임시 저장 공간이라고 생각하면 이해하기 쉽다.
티켓팅 서버를 예시로 들면 유저 트래픽이 갑작스레 증가 할 경우 서버가 터질 수 있다. 이 때 Message Queue를 통해 유저 인증 절차를 순차적으로 하게 된다면, 접속 순서에 따라 인증이 순차적으로 진행될 것이고, 인증이 완료된 사람은 티켓팅을 진행할 수 있다.

hello가 Message Queue가 담겨있는 저장 공간이다.
P는 Producer로써 MQ에 데이터를 저장 한다. MQ에는 데이터가 순차적으로 저장되고, C는 Consumer 즉 소비자로써, MQ에 저장된 데이터를 순차적으로 읽어 들인다.

Install

RabbitMQ는 Docker이미지로도 존재한다. Mac OS의 경우 brew로도 쉽게 설치할 수 있지만, 오늘은 Docker를 통해 간편히 실행시켜 볼 것이다.

$ docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.12-management

해당 명령어를 터미널에 입력 하면 RabbitMQ 이미지를 다운 받은 뒤, 컨테이너가 실행될 것이다.

Node에서 RabbitMQ를 실행시키기 위해선 amqp라이브러리를 설치해야 한다. 개방형 메시지 프로토콜인 amqplib 라이브러리를 설치 해 준다.

$ npm install amqplib

실행

Message Queue를 테스트 하기 위해선, Sender와 Reciver 두개의 파일이 존재 해야 한다. 두개의 파일을 통해, Message Queue를 테스트 해볼 것이다.

Sender

import amqp from "amqplib/callback_api";

amqp.connect("amqp://localhost", function (error0, connection) {
  if (error0) {
    throw error0;
  }
  connection.createChannel(function (error1, channel) {
    if (error1) {
      throw error1;
    }
    const queue = "hello";
    const msg = "Hello world";

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

    for (let i = 0; i < 30; i++) {
      channel.sendToQueue(queue, Buffer.from(msg + i));
    }
    console.log(" [x] Sent %s", msg);
  });
  setTimeout(function () {
    connection.close();
    process.exit(0);
  }, 500);
});

먼저 ampq.connect를 통해 RabbitMQ에 접속한다. 그 후, createChannel메서드를 통해 hello라는 채널을 생성 한뒤, 마지막으로 sendToQueue 메서드를 통해 Queue에 메시지를 저장할 수 있다.

위 코드를 실행시킨 뒤 RabbitMQ의 저장공간을 살펴 보면, for문을 통해 생성된 30개의 메시지를 확인 해 볼 수 있다.

$ docker exec -it rabbitmq /bin/bash

위 명령어를 통해 docker rabbitmq 컨테이너로 진입한다.

$ rabbitmqctl list_queues

hello라는 채널 명에 messages가 30개 생성된 모습

Reciver

import amqp from "amqplib/callback_api";

amqp.connect("amqp://localhost", function (error0, connection) {
  if (error0) {
    throw error0;
  }
  connection.createChannel(function (error1, channel) {
    if (error1) {
      throw error1;
    }
    const queue = "hello";

    channel.assertQueue(queue, {
      durable: false,
    });
    console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue);

    channel.consume(
      queue,
      function (msg) {
        if (msg !== null) {
          console.log(" [x] Received %s", msg.content.toString());
        }
      },
      {
        noAck: true,
      }
    );
  });
});

Reciver또한 RabbitMQ에 접속 후 채널을 생성하기 까진 동일하게 작동 한다.
Message Queue에 저장된 메시지를 수신하기 위해서는 consume 메서드를 사용한다. consume메서드를 이용해 console로 해당 메시지를 출력 하도록 코드를 작성 했다.

해당 코드를 실행 시키면 Sender에서 Message Queue에 저장해 두었던 30개의 메시지가 출력 될 것이다.

이 처럼 Message Queue를 이용하면, 송신 서버와 수신 서버가 동시에 작동하지 않더라도, 작동이 보증되어 시스템의 안정성이 늘어난다.
가령 Serverless를 구현했을 경우, Cold Start라는게 존재해 데이터가 유실될 수 있는데, Message Queue를 이용하면, 서버가 작동 후 로직이 진행되기 때문에 데이터의 유실을 막을 수 있다.

마치며

RabbitMQ의 예제코드와 단순 실행을 통해 Message Queue에 대해 간단히 알아 보았다. 다음 게시글에서는 Message Queue를 통해 더욱 심화된 내용을 다뤄 볼 것이다.