Повышаем производительность Ruby on rails приложений с помощью ActiveMQ

В своём посте хочу рассказать о возможности использования ActiveMQ в проекте написанном на фреймворке Ruby on rails.

Что такое Message Queue?

MQ – это архитектура обмена сообщениями между компонентами приложения в асинхронном режиме. Т. е. отправитель и получатель могут взаимодействовать в разное время. Состоят такие системы из producer'а (отправителя) и consumer'a (получателя) которые взаимодействуют между собой через broker.

Используя такие системы можно существенно увеличить производительность приложения, выполняя код в асинхронном режиме. Допустим у вас есть код который очень замедляет выполнение какой то части на вашем сайте, чтобы пользователь не ждал завершение работы такого кода, лучше его выполнить в асинхронном режиме. Несколько простых примеров:
— генерация thumbnails;
— сбор статистики;
— рассылка писем/сообщений;
— удаление данных с таблиц;
— индексация данных;
— импорт данных в базу.

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

Что такое ActiveMQ?

ActiveMQ — это открытая реализация message broker системы. Преимущество данной системы это высокая производительность, открытость и возможность реализации клиентов на любых языках. ActiveMQ в данный момент поддерживает следующее протоколы:
— OpenWire (собственный бинарный протокол);
— Stomp;
— REST;
— WS Notification;
— XMPP.

Установка ActiveMQ

Описание будет идти для Linux, хотя думаю проблем с установкой на Windows не должно быть. Выполняем следующие команды:

wget apache.multihomed.net/activemq/apache-activemq/5.2.0/apache-activemq-5.2.0-bin.tar.gz
tar xvf apache-activemq-5.2.0-bin.tar.gz
sudo cp -R ./apache-activemq-5.2.0 /usr/local/apache-activemq

Установка завершена. Запускаем ActiveMQ:

sudo /usr/local/apache-activemq/bin/activemq &

Все готово к работе.

Работа с ActiveMQ на Ruby on rails

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

rails test
cd test
script/generate controller test
script/server

Преобразуем код контроллера Test к следующему виду:

class TestController < ApplicationController
  def index
    self.class.benchmark('Write to file') do
      path = File.join(RAILS_ROOT, 'tmp', 'test')
      file_test = File.new(path, 'w')

      10000000.times do
        file_test.write("Some test message\n")
      end
     end

    render :text => 'Data saved'
  end
end

Открываем http://0.0.0.0:3000/test/index и видим, что приходиться какое-то время ждать завершение работы кода. У меня бенчмарк показал, что данный код выполняется 10 секунд, очень много, не каждый пользователь готов ждать такое время.

Для взаимодействия с ActiveMQ я предлагаю использовать плагин ActiveMessaging, хочу отметить, что есть возможность использовать непосредственно Stomp клиент. Но в таком случае мы потеряем возможность легко и просто переключиться на другую систему MQ.

ActiveMessaging — это плагин для Ruby on rails который очень легко использовать в проекте для выполнения кода в асинхронном режиме. Поддерживает большое количество протоколов, в том числе и Stomp, т.е. мы можем его использовать для работы с ActiveMQ.

Устанавливаем его в проект:

script/plugin install activemessaging.googlecode.com/svn/trunk/plugins/activemessaging

также нам нужно установить клиент для работы с stomp протоколом:

gem install stomp

Установка завершена.

ActiveMessaging имеет два конфигурационных файлов:
— config/broker.yml — содержит настройки для подключения;
— config/messaging.rb — содержит настройки очередей.

Выполняемый код в асинхронном режиме пишется в так званных процессорах. Для генерации нашего первого процессора выполним следующую команду:

script/generate processor Test

Открываем наш процессор (/app/processors/test_proccesor.rb) и переносим сюда тестовый код:

class TestProcessor < ApplicationProcessor
  subscribes_to :test

  def on_message(message)
    path = File.join(RAILS_ROOT, 'tmp', 'test')
    file_test = File.new(path, 'w')

    10000000.times do
      file_test.write("#{message}\n")
    end

    logger.debug 'Data saved'
  end
end

Преобразуем код экшена index на следующее:

def index
  publish :test, 'Some test message'
  render :text => 'Message sent'
end

Здесь мы просто передаём сообщение 'Some test message' в destination под именем test, т. е. выполняем функцию producer'а. Если вы откроете конфигурационный файл config/messaging.rb то вы увидите под именем destination'а test очередь /queue/Test, в которую наше сообщение и будет отправлено.

Запускаем демон, который является у нас consumer'ом, т.е. он будет читать все сообщения с очереди /queue/Test и передавать их процессору TestProcessor.

script/poller start

Для чистоты теста удаляем наш файл в который мы писали данные:

rm /tmp/test

Обновляем страницу http://0.0.0.0:3000/test/index и сразу видим, что код выполняется практически моментально, а теперь идем в tmp и видим вновь созданный файл test с данным. Также можно перейти в лог и там увидеть сообщение 'Data saved'.

В большинстве случаев нам нужно выполнить код с какими то параметрами. Мы можем перевести хеш с параметрами в XML, JSON или Yaml и передать в виде сообщения. Я предпочитаю JSON.

Давайте попробуем передать количество записываемых строк и само сообщение, которое записывается в файл. Немного изменим экшен index:

def index
  publish :test, {:count => 200000, :message => 'Some new test message'}.to_json
  render :text => 'Message sent'
end

Изменим и метод on_message:

def on_message(message)
  options = JSON.parse(message).symbolize_keys
  path = File.join(RAILS_ROOT, 'tmp', 'test')
  file_test = File.new(path, 'w')

  options[:count].times do
    file_test.write("#{options[:message]}\n")
  end

  logger.debug 'Data saved'
end

Так как в ruby нет стандартных функций для работы с JSON, нужно установить gem:

sudo gem install json

и подключить его в TestProcessor:

require 'json'

Теперь прерываем работу нашего демона и запускаем заново:

script/poller stop
script/poller start

Снова обновляем страницу http://0.0.0.0:3000/test/index. Открыв файл tmp/test, мы увидим перезаписанный файл уже строками 'Some new test message'.

В процессоре очень легко использовать модели, точно также как и в обычных экшенах. Небольшой пример для наглядности:

def on_message(message)
  options = JSON.parse(message).symbolize_keys

  Product.find(:all).each do |product|
    product.price = options[:price]
    product.save
  end
end

В ActiveMQ существует утилита, в которой вы сможете смотреть статистику отосланных сообщений, список очередей и даже отослать сообщение, очень удобно использовать для отладки. Находиться она по адресу http://0.0.0.0:8161/admin/.

Надеюсь, этот пост продемонстрировал простоту и преимущество использования ActiveMQ в проекте. Всего-навсего небольшой рефакторинг в вашем проекте и приложение задышит по другому.