Load Balancing: Firebase + RabbitMQ

Современные хостинг-системы (Heroku, Amazon etc.) предоставляют широкий выбор устройств и настроек для проектирования архитектуры балансировки нагрузки на сервер. Вы можете настроить как более простой Round Robin-алгоритм для последовательной разгрузки сервера, так и более сложную систему, учитывающую количество instanc'ов, текущую нагрузку, среду окружения и другие факторы.

Сегодня мы поговорим о ручном способе регулирования нагрузки (одним из). Сразу скажу, что данный способ не был протестирован в жестких условиях, но достаточно хорошо показал себя в pre-production.

Так же следует добавить, что способ, о котором я расскажу, не является каким-то лекарством или заменой уже предоставляемым хостинг-платформами balancer'ов, но в ситуации урезанного бюджета может являться этаким подорожником.

Итак, приступим. Что нам необходимо:
— Firebase (Cloud NoSQL Database)
— Message Broker. В данном случае RabbitMQ

Теория

Firebase

Тема Firebase уже была несколько раз задета раннее:
Создание AngularJS приложения c использованием Firebase
Авторизация пользователей с AngularJS и Firebase
В одной упряжке Polymer’ы, Dart и Firebase

Но в рамках этой статьи я расскажу те вещи, которые необходимо знать для ее понимания.

Listeners
На данный момент Firebase предоставляет несколько методов для «прослушивания» событий, происходящих в базе данных. Вы можете с легкостью узнать, когда определенное поле было изменено, удалено, создано или перемещено. Для этого могут быть использованы методы on() и once(). Различаются они только тем, что on() будет срабатывать каждый раз, как следствие, проводимых с базой действий, в то время как once() срабатывает единожды.

Примеры:

var Firebase = require('firebase');
var dbConnectionUrl = 'https://foo-database.firebaseio.com/posts/java_script'
var dbReference = new Firebase(dbConnectionUrl);

dbReference.on('child_added', function(addedChildSnap){
    var addedChildValue = addedChildSnap.val(); //addedChildSnap содержит в себе т.н. "слепок", а не только значение 
    doDomethingWithAddedChild(childValue);
});

dbReference.once('child_removed', function(removedChildSnap){
    var removedChildValue = removedChildSnap.val();
    doSomethingWithRemovedChild(removedChildValue);
});

События «child_added» и «child_removed» являются частью Firebase API (я не выдумал их из головы). Для ознакомления с ними подробнее можно пойти по ссылкам:
www.firebase.com/docs/web/api/query
www.firebase.com/docs/web/api/query/on.html

«child_added» будет срабатывать каждый раз при добавлении нового post'а в «java_script» (обратите внимание на dbConnectionUrl) т.к. мы использовали on()-слушатель. В случае с child_removed («выстреливает» при удалении сущности) мы используем once(), поэтому событие отработает один раз и, если мы удалим больше одного элемента, то никаких действий предпринято не будет.

RabbitMQ

Опять же, данная технология уже рассматривалась на Хабре.
RabbitMQ tutorial 1 — Hello World
…
RabbitMQ tutorial 6 — Удаленный вызов процедур

Т.е. это такой себе почтальон, целью которого является принимать и отдавать сообщения. Эти сообщения могут накапливаться и храниться в нем до полной их отправки. В контексте этой статьи стоит знать, что этот почтальон может отдавать сообщение одно за одним, не зависимо от того, удачно ли их принимает слушатель или же отдавать новое сообщение только тогда, когда принимающая сторона «отстучалась» ему и сказала, что сообщение получено, мол, спасибо. Так вот, в нашем случае нам необходим второй способ для правильной работы.

Общая схема

Front-End не работает c Back-End'ом вообще
Специфика Firebase такова, что работать с данной базой можно не использую серверную часть, что дает некоторые преимущества в скорости.

Front-End вносит изменения в базу — Back-End «слушает» эти изменения
Использую стандартные средства Firebase мы можем с легкостью «прослушивать», проводимые в базе данных действия.

Back-End посылает сообщения в очередь
Получив отклик от Firebase как результат, проведенного над базой действия, back-end формирует объект и посылает его в очередь

RabitMQ Server отдает сообщения
Данные сервис отвечает за хранения и «выдачу» сообщений по мере их появления в очереди

Практика

Условия
Database Url: https://foo-database.firebaseio.com/
Database Entities: users_approved, users_requested
(https://foo-database.firebaseio.com/users_approved, https://foo-database.firebaseio.com/users_requested)

Задача
Front-End будет создавать сущности в базе данных (users_requested), в свою очередь Back-End будет проверять объекты на валидность и выполнять последующую магию.

Я не буду создавать полнофункциональный Front-End, дабы не засорять статью лишним кодом, поэтому просто покажу только необходимые функциональные части. Это, кстати, касается и Back-End части.

Front-End

var dbRootReference = new Firebase('https://foo-database.firebaseio.com/');
var newRequestedUser = {
    firstName: 'John',
    lastName: 'Connor',
    email: connor.john@domain.com
};

dbRootReference.child('users_requested').push(newRequestedUser);

Этого достаточно для того, чтобы добавить новую запись в «requested_users» с клиентской части.
Само собой я пропустил шаг с аутентификацией в базу данных. Не стоит думать, что тут все очень плохо с безопасностью. Подробнее с auth-методом можно ознакомиться здесь

Back-End: RabbitMQ

//подключаем библиотеку для работы с AQMP протоколом - стандарт, через который работает RabbitMQ
//есть еще масса других библиотек (amqp, node-amqp, bramqp и т.д.), но в данном случае я использую "amqplib"
var amqp = require('amqplib');
var amqpConnectionUrl = 'amqp://localhost';
//создаем подключение
var amqpConnection = amqp.connect(amqpConnectionUrl);

//для ясности инициализируем константы тут
var QUEUE_USERS_REQUESTED = 'requestedUsers';
var QUEUE_USERS_APPROVED = 'approvedUsers';

//создаем слушателя (именно он будет принимать сообщения)
amqpConnection.then(function (successAmqpConnection) {
  successAmqpConnection
    .createChannel()
    .then(function (amqpChannel) {
      //в данном примере я прикручиваю обе очереди к одному каналу
      //в дальнейшем вы можете выделять отдельный канал для своей очереди, в зависимости от нагрузки и целей использования

      //прикрепляем в channel нашу очередь для requested пользователей
      amqpChannel.assertQueue(QUEUE_USERS_REQUESTED, {durable: false, noAck: false});  
      //прикрепляем в channel нашу очередь для approved пользователей
      amqpChannel.assertQueue(QUEUE_USERS_APPROVED, {durable: false, noAck: false});  
      //noAck опция является как раз тем параметром, который должен быть выставлен в TRUE (ack: true или noAck: false),
      //если мы хотим, чтобы новое сообщение отдавалось только тогда, когда предыдущее обработано полностью

      //слушатель для очереди QUEUE_USERS_REQUESTED
      amqpChannel.consume(QUEUE_USERS_REQUESTED, function (msg) {
        if (msg !== null) {
          var requestedUserDetails = JSON.parse(msg.content.toString()); //мы ждем JSON-объект пользователя
          return removeUserFromRequests(requestedUserDetails)
            .then(validateUser)
            .then(function (isValid) {
              if (isValid) {
                return;
              } else {
                throw Error('user not valid');
              }
            })
            //создаем запись в 'users_approved'
            //dbRootReference.child(USERS_APPROVED).push(requestedUserDetails);
            .then(addUserAsApproved)
            .then(function (result) {
              //Не забываем сказать сервису, что сообщение принято и обработано с помощью метода ack();
              amqpChannel.ack(msg); //сообщаем сервису, что сообщение получено, можно слать следующее
            })
            .catch(function (err) {
              //обрабатываем ошибку и сообщаем серверу о полученном сообщении
              amqpChannel.ack(msg);
            });
        } else {
          //пишем обработку пустых сообщений
          //и сообщаем сервису, что сообщение получено, можно слать следующее - не стоит засорять стек
          amqpChannel.ack(msg); 
        }
      });

      //слушатель для очереди QUEUE_USERS_APPROVED
      amqpChannel.consume(QUEUE_USERS_APPROVED, function (msg) {
        if (msg !== null) {
          var approvedUserDetails = JSON.parse(msg.content.toString()); //мы ждем JSON-объект пользователя
          sendWelcomeMessage(approvedUserDetails) //шлем пользователю письмо
            .then(function (result) {
              amqpChannel.ack(msg); //уведомляем AQMP-сервис, что сообщение принято и можно слать следующее
            });
        } else {
          amqpChannel.ack(msg); //сообщаем сервису, что сообщение получено, можно слать следующее
        }
      });

    });
});

//создаем publisher'ов для отправки сообщений в очередь 
function pushToUserRequestedQueue(requestedUser) {
  return amqpChannel.sendToQueue(QUEUE_USERS_REQUESTED, JSON.stringify(requestedUser));
}

function pushToUserApprovedQueue(approvedUser) {
  return amqpChannel.sendToQueue(QUEUE_USERS_APPROVED, JSON.stringify(approvedUser));
}

Back-End: Firebase

//устанавливаем соединение 
var Firebase = require('Firebase');
var dbUrl = 'https://foo-database.firebaseio.com/';
var dbRootReference = new Firebase(dbUrl);

var USERS_REQUESTED = 'users_requested';
var USERS_APPROVED = 'users_approved';

//слушаем 'users_requested' на предмет новых записей
dbRootReference.child(USERS_REQUESTED).on('child_added', function (requestedUserSnap) {
  var requestedUser = requestedUserSnap.val(); 
  //отправляем сообщение в очередь запрошенных пользователей
  sendMessageToUserRequestedQueue(requestedUser);
});

//слушаем 'users_requested' на предмет новых записей
dbRootReference.child(USERS_APPROVED).on('child_added', function (newApprovedUserSnap) {
  var approvedUser = newApprovedUserSnap.val();
  //отправляем сообщение в очередь подтвержденных пользователей
  sendMessageToUserApprovedQueue(approvedUser);
});

Заключение

Вот, собственно, и все. При обнаружении данных в базе по пути 'users_requested', наш сервер отправит сообщение в очередь, которое дальше будет обработано, исходя из бизнес-логики, внедренной в проект. Новое сообщение будет обработано только тогда, когда на это будет дано разрешения, тем самым снижая нагрузку на сервер. Отправка сообщение в очередь требует гораздо меньше ресурсов, нежели, к примеру, провести ряд операций для валидации и сохранения пользователя.

Pros & Cons

— используя именно Firebase в данном случае мы уменьшаем время отклика базы данных, а так же скорость обработки данных
— получаем бюджетный вариант load-balancer'а на этапе создания MVP
— отсутствие Map Reduce в Firebase делает ее сомнительной в работе с большими объемами данных, с другой стороны Real-Time принцип очень хорошо себя показывает, поэтому с правильным подходом все может получиться достаточно качественно
— ввиду того, что для Firebase еще нет удобной ORM (как я считаю), базой не совсем удобно пользоваться
— не каждое приложение может быть совместимо с такой схемой, скажем, ждать когда персональные данные пользователя будут изменены по его запросу не очень хорошо, но в случае обработки заказов интернет-магазина она вполне подходит
— доступное API и плотная не размытая информация вносит приятные моменты в изучение и разработку