1. node.js로 rabbitmq 쓰기 - Send & Receive

개요

이번 포스팅에서는 node.js로 메시지 큐 중에 하나인 rabbitmq를 사용하는 법에 대해서 다루도록 하겠습니다.

What is Rabbitmq?

Rabbitmq는 메시지 브로커 서비스로, 불특정 다수의 사용자가 다른 사람에게 메시지를 전달하려고 메시지 브로커에 메시지를 전달하면 메시지 브로커가 메시지를 받아서 보관하고 있다가 해당 메시지를 받으려는 사람이 보관한 메시지를 꺼내가는 형태의 서비스 입니다.

한마디로, 우체통이나 편지함을 생각해보면 쉽습니다. 편지를 보내려는 사람이 편지를 써서 우체통에 넣으면 받는 사람이 우체통에서 편지를 꺼내가는 구조이지요.

즉 정리하면 아래와 같습니다.

• 보내는 사람 = 메시지를 보내는 사람
• 받는 사람 = 메시지를 꺼내가는 사람
• 우체통 = 메시지 브로커

보통 RabbitMQ와 같은 메시지 브로커는 큐 기반의 자료구조를 기반으로 하고 있습니다. 보내는 사람이 메시지를 보내서 메시지큐에 쌓아두면, 받는 사람이 해당 메시지 큐에서 메시지를 꺼내간 뒤 그 내용에 따라 적절한 작업을 수행하는 방식으로 보통 진행됩니다.

실제로도 많은 비동기 라이브러리, 서비스, 심지어 인터럽트와 같은 거의 대다수의 비동기적으로 동작하는 모든 코드들이 이와 같은 구조를 따르고 있습니다.

사실 rabbitmq 이외에도 kafka, zeromq, activemq와 같은 많은 종류들의 메시지 큐가 있습니다. (심지어 redis와 같은 데이터베이스도 메시지큐로 사용할 수 있습니다.)

node.js에서는 rabbitmq를 사용하기 위해 amqplib이라는 라이브러리를 사용하고있습니다.

이번시간에는 node.js와 rabbitmq를 이용해서 간단한 생산자, 소비자 예제를 작성하도록 하겠습니다.

설치

먼저 아래 명령어를 이용해서 rabbitmq를 설치해줍니다.

brew install rabbitmq

그런뒤, 아래 명령어를 이용해서 rabbitmq 서비스를 시작해줍니다.

brew services start rabbitmq

마지막 으로, 아래 명령어를 이용해서 node.js에서 사용하는 amqplib을 설치해 줍니다.

npm install amqplib

이로써 기본적인 환경설정 세팅이 끝나게 되었습니다.

예제

이번 예제는 rabbitmq를 이용해서 생산자, 소비자 예제를 구현하는 예제입니다.

생산자(send.js)가 계속 데이터를 생성해서 메시지 브로커를 통해 메시지 큐에 집어넣으면은 소비자(recv.js)가 메시지 브로커를 통해 메시지를 가져와서 출력하는 예제입니다.

아래 예제는 생산자에 해당하는 send.js 코드 입니다.

#!/usr/bin/env node

var amqp = require('amqplib/callback_api');

amqp.connect('amqp://localhost', function(error0, connection) {
    if (error0) {
        throw error0;
    }
    connection.createChannel(function(error1, channel) {
        if (error1) {
            throw error1;
        }

        var queue = 'hello';
        var msg = 'Hello World!';

        channel.assertQueue(queue, {
            durable: false
        });
        channel.sendToQueue(queue, Buffer.from(msg));

        console.log(" [x] Sent %s", msg);
    });
    setTimeout(function() {
        connection.close();
        process.exit(0);
    }, 500);
});

위의 코드는 먼저, connect함수를 이용해서 rabbitmq 메시지 브로커에 연결을 수행합니다. 이때 메시지를 주고받는 프로토콜로 "amqp"라는 프로토콜을 선택합니다. (사실 저거 이외에도 http와 같은 다양한 프로토콜을 선택할수 있습니다.)

다음으로 createChannel 함수를 통해서 채널을 생성합니다. 채널은 나중에 다시 설명을 하겠지만, 지금은 일단 송신자와 수신자를 잇는 일종의 파이프 비슷한거라고 생각하시면 되겠습니다.

채널 생성이 완료되면 assertQueue 함수를 이용해서 큐를 생성(없는경우는 새로 생성, 있는 경우는 해당 큐를 선택)해주고, sendToQueue 함수를 이용해서 선택된(즉, 선언된) 메시지 큐에 메세지를 전송합니다. 그런뒤 연결을 종료한뒤에 프로그램을 종료합니다.

아래 예제는 소비자(receive.js) 예제입니다.

#!/usr/bin/env node

var amqp = require('amqplib/callback_api');

amqp.connect('amqp://localhost', function(error0, connection) {
    if (error0) {
        throw error0;
    }
    connection.createChannel(function(error1, channel) {
        if (error1) {
            throw error1;
        }

        var 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) {
            console.log(" [x] Received %s", msg.content.toString());
        }, {
            noAck: true
        });
    });
});

send.js와 유사하게 connect함수를 이용해서 rabbitmq 메시지 브로커에 연결을 수행한뒤, 채널을 생성하고 큐를 선택 해주고, sendToQueue 함수를 이용해서 선택된 메시지 큐에서 메시지를 꺼내서 출력을 해줍니다. 이때, noAck 옵션을 켜서 메시지를 받았다는 응답은 하지 않도록 설정하였습니다.

위 코드를 전부 작성 하면 아래와 같이 실행 할 수 있습니다.

./send.js

./receive.js

실행하면, receive.js에서 send.js에서 보낸 메시지가 출력이 되는것을 알 수 있습니다. (Receive.js는 Cmd+C를 눌러서 종료할 수 있습니다.)

위의 예제를 보면 알겠지만, 소켓 통신처럼 데이터를 주고받는것과 유사하다는 것을 알 수 있습니다. 실제로 메시지 브로커를 소켓 통신과 유사한 목적으로도 쓸 수 있습니다.