4. node.js와 rabbitMQ 에서의 Routing

개요

이번 포스팅에서는, node.js를 이용해서 rabbitMQ를 사용할때 메시지를 라우팅 하는 방법에 대해서 다루어 보도록 하겠다.

라우팅?

라우팅이란 단어를 처음 들어본 사람들도 있을것이고, 네트워크 시간에 간략하게나마 들어본 사람도 있을것이다.

라우팅이란 단어를 사전에서 찾아보면 아래와 같은 뜻이 나온다.

라우팅(영어: routing)은 어떤 네트워크 안에서 통신 데이터를 보낼 최적의 경로를 선택하는 과정이다. 최적의 경로는 최단 거리가 될수도 있으며, 가장 빠른 시간에 전달하는 것으로 계산될수도 있다. 라우팅은 전화 통신망, 전자 정보 통신망(인터넷 같은) 그리고 교통망 등 여러 종류의 네트워크에서 사용된다

즉 핵심은, 어떠한 "데이터 덩어리"가 목적지 까지 최단(최적) 경로로 길을 찾아가는 것이다.

이 말을 rabbitMQ에 맞춰서 고쳐쓰면 아래와 같이 된다.

라우팅 : 네트워크(즉, 전체 서비스) 안에서, 메시지를 어느 큐에 보낼지를 선택하는 과정

즉, 메시지를 어느 큐에 보낼지를 고르는 과정이라고 볼 수 있다.

이제 대충 어느 개념인지 이해는 갔으니 예제를 보도록 하자.

예제

지난 포스팅에서, rabbitMQ로 pub-sub 패턴을 구현할때, 바인드(bind) 라는 개념을 배웠을 것이다.

사실 바인딩은 메시지 브로커의 익스체인지(exchange) 와 메시지 큐를 이어주는 작업이다.

익스체인지(exchange)는 메시지 브로커에서 메시지를 어느 큐에 보낼지를 결정하는 작업이다.

그리고 rabbitMQ에서 바인딩 작업을 수행할시, 메시지의 키값이 무엇이냐에 따라 어떤 큐에 넣을지를 결정 할 수 있다.

즉 익스체인지(exchange)와 바인딩(binding) 기능이 바로 rabbitMQ에서의 라우팅의 핵심이다.

또한, 키 1개로 여러개의 큐에 바인딩을 시킬 수도 있다.

예를 들면 "info","warning","error"라는 3가지 종류의 로그를 계속해서 찍어내고 있는 프로그램이 있다고 가정해보자.

라우팅을 이용하면 "info", "warning", "error"에 해당하는 각 종류의 로그를 분리해서 메시지 큐에 보냄으로써, 로그 분류 시스템을 만들수 있는것이다.

이러한 특성들을 이용해서, 심각도에 따라 로그를 분류하는 시스템을 rabbitMQ로 만들어 보자.

아래 예제는, rabbitMQ를 통해서 로그의 종류를 지정하고 지정한 종류의 로그를 보내는 예제이다.

#!/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 exchange = 'direct_logs';
    var args = process.argv.slice(2);
    var msg = args.slice(1).join(' ') || 'Hello World!';
    var severity = (args.length > 0) ? args[0] : 'info';

    channel.assertExchange(exchange, 'direct', {
      durable: false
    });
    channel.publish(exchange, severity, Buffer.from(msg));
    console.log(" [x] Sent %s: '%s'", severity, msg);
  });

  setTimeout(function() { 
    connection.close(); 
    process.exit(0) 
  }, 500);
});

위 코드에서, connect, createChannel 함수를 이용해서 채널 생성후, publish 함수에 어떤 익스체인지(exchange)를 사용할지와 키 값(여기서는 severity)를 정해서 보내는 것을 알 수 있다.

아래 예제는, rabbitMQ를 이용하여 사용자가 지정한 종류들의 로그를 받아보는 예제이다.

#!/usr/bin/env node

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

var args = process.argv.slice(2);

if (args.length == 0) {
  console.log("Usage: receive_logs_direct.js [info] [warning] [error]");
  process.exit(1);
}

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

    channel.assertExchange(exchange, 'direct', {
      durable: false
    });

    channel.assertQueue('', {
      exclusive: true
      }, function(error2, q) {
        if (error2) {
          throw error2;
        }
      console.log(' [*] Waiting for logs. To exit press CTRL+C');

      args.forEach(function(severity) {
        channel.bindQueue(q.queue, exchange, severity);
      });

      channel.consume(q.queue, function(msg) {
        console.log(" [x] %s: '%s'", msg.fields.routingKey, msg.content.toString());
      }, {
        noAck: true
      });
    });
  });
});

위 코드에서는, bindQueue 함수를 통해 메시지를 받아볼 익스체인지(exchange)와 키 값(여기서는 severity)를 이용해서 메시지를 받아볼 큐를 정하고 consume 함수를 이용해서 받아보는것을 확인 할 수 있다.