Gearman — сервер очередей: использование в проектах на PHP

Не так давно была замечательная статья, описывающая общие принципы работы с сервером очередей Gearman. Мне бы хотелось продолжить материал, дополнив его некоторыми деталями практического применения, а именно:
— установка и управление сервером
— управление очередью — что возможно и как
— PECL и PEAR php-расширения для работы с Gearman
— мониторинг сервера
— примеры кода
— передача данных порциями
— организация параллельных вычислений в PHP

Интересно? Прошу под кат

Прежде чем приступить к изложению материала, нужно сказать, откуда он появился. Это компиляция из практического применения в нескольких проектах (все касались работы с медленными удаленными сайтами), выдержек из документаций и обсуждений, просто некоторых наблюдений и абстрактных соображений. Разумеется, могут быть неточности, спорные решения и спорные мысли — автор будет благодарен за любую критику и правки.

Содержание

Жизненный цикл данных Gearman
Установка и управление Gearman
Мониторинг состояния сервера

Cудьба очереди при рестарте gearman
Сброс всей очереди и сброс очереди конкретной задачи
Перезапуск воркера

PECL и REAR php-расширения для работы с Gearman
pear Net_Gearman
pecl gearman
Установка pecl gearman

Самый простой клиент и воркер (+видео)
Передача данных по частям и дополнительный обмен данными с клиентом(+видео)
Фоновые задачи и приоритет
Параллельные вычисления в PHP с использованием Gearman(+видео)

Жизненный цикл данных Gearman

Прежде чем вдаваться в подробности, еще раз проиллюстрируем работу с сервером.
Самый простой пример. Как происходит вызов и выполнение ф-и в PHP:



Тривиально. Теперь то же самое, но с использованием сервера очередей:



Вкратце словами: скрипт, которому требуются результаты/действия работы функции (“client”) отправляет на сервер (регистрирует) имя функции и ее аргументы — на сервере создается задача (“task”).

Если на сервере зарегистрирован обработчик для функции с таким именем (“worker”) и он в данный момент свободен, ему передаются данные для обработки и имя ф-и в виде задачи (“job”).
Если worker для такой ф-и не зарегистрирован или он занят, task становится в очередь и ждет обработки.

После обработки worker передает то, что возвращает ф-я, обратно на Gearman. Сервер очередей смотрит, какой client регистрировал task с такими именем ф-и и с такими данными, и отправляет результат работы worker ему (это в том случае, если клиент не регистрировал задачу как фоновую. Если задача зарегистрирована как фоновая, клиенту ничего не передается).

Что сразу следует из такого принципа работы?

1. В работе с Gearman участвуют четыре объекта: client, task, worker, job. Описание всех объектов есть на оф. сайте PHP.

2. Передаваемые данные — и к серверу от клиента, и обратно от воркера — это только строка, и только одна. Если требуется передать несколько аргументов или не строку — массив к примеру — требуется сериализация.

Впрочем, все изложенное выше — просто краткое повтореное оф. документации и уже опубликованных материалов. Теперь нюансы.

Установка и управление Gearman

Если есть большая необходимость, можно запустить сервер Gearman на Windows, есть его реализация на Java:
https://launchpad.net/gearman-java

Но мы будем рассматривать работу под linux (нижеследующие примеры приведены для debian).
Установка сервера проходит гладко и особенностей не имеет:

aptitude install gearman-job-server

После успешной установки сервер управляется из etc/init.d/gearman-job-server простыми и ясными командами:
{ start | stop | restart | force-reload }
Хост по умолчанию для сервера — localhost, порт — 4730

Мониторинг состояния сервера

Всегда есть необходимость посмотреть, что происходит на сервере: какие задачи зарегистрированы, сколько их в очереди, сколько воркеров зарегистрировано на каждую задачу.
Это можно сделать из консоли командой
(echo status) | netcat 127.0.0.1 4730

Cудьба очереди при рестарте gearman

Сразу возникает вопрос: что будет с очередью на работающем сервере, если сделать ему харакири restart?
При дефолтной установке очередь хранится в памяти, рестарт сервера аннулирует все задачи, которые клиенты отправили на сервер, и аннулирует регистрацию всех воркеров на ��ервере. При этом у воркеров сгенерируются исключения вроде «Потеряна связь с сервером».
Очередь можно сохранить в БД, поддерживаются MySQL, PostgreSQL, SQLite. Как организовать такую очередь, хорошо написано на оф. сайте:
http://gearman.org/index.php?id=manual:job_server#persistent_queues

Следует иметь ввиду, что даже если сохранить очередь, при рестарте сервера обработчики задач — воркеры — все равно «отвалятся» от сервера, и придется либо их перезапускать, либо заранее в самих воркерах такую ситуацию предусматривать.

Сброс всей очереди и сброс очереди конкретной задачи

Если очередь не хранится в БД, рестарт сервера, как сказано выше, сбросит всю очередь. Если очередь хранится в БД, нужно, кроме рестарта сервера, очистить соответствующую таблицу — всю или выборочно. Напомним, что рестарт сервера сгенерирует исключения как в клиентах, так и в воркерах.

Однако может возникнуть ситуация, когда нужно сбросить очередь только по одной задаче. Пример: мы парсим 100 сайтов, один перестал отвечать, в очереди накопилось 1000 задач по приему данных с этого сайта, данные со всех остальных сайтов принимаются через сервер очередей без проблем. Нужно сбросить очередь только по заглохшему сайту.
Самый простой и безболезненный способ это сделать — запустить фейковый воркер, который зарегистрирует на сервере очередей задачу с таким же именем, но быстро возвращающую NULL, пустую строку или вообще все что угодно — главное быстро. Этот воркер пропустит через себя все ожидающие задачи, очередь очистится.

Перезапуск воркера

Для чего это может потребоваться? Ну кроме очевидной ситуации, когда воркер завис и это не вызывает сомнений, есть еще одна, очень частая.
Ситуация: воркер запущен, зарегистрировал на сервере свои задачи. Если вы в этот момент внесете изменения в код воркера, то сколько бы вы его не сохраняли, обрабатывать задачи с сервера очередей будет тот код, который был в момент регистрации задачи.
Для того, чтобы изменения в коде вступили в силу — то есть чтобы задачи с сервера обрабатывал уже измененный код, воркера надо перезапустить, то есть прекратить выполнение скрипта текущего воркера и запустить его заново. Способы тривиальны: вручную, bash-скриптом, предусмотреть в воркере обработку ф-и вроде exit_worker и запуск его из крона etc.

PECL и REAR php-расширения для работы с Gearman

Для работы с сервером очередй Gearman в php можно использовать два варианта расширений: pecl gearman и pear Net_Gearman

Между этими расширениями есть принципиальная разница.

pear Net_Gearman

pear Net_Gearman — это просто несколько php-файлов. И все. Установить на сервер очень просто:
pear install Net_Gearman channel://pear.php.net/Net_Gearman-0.2.3
Можно даже на сервер не устанавливать — просто распаковать архив, реализовать подключение соотв. классов и все — можно использовать, вот так например

function __autoload($className){

    //Костыль для pear Net_Gearman, кроме Net_Gearman_Job
    if(strstr($className, 'Net_Gearman_') and !strstr($className, 'Net_Gearman_Job_')){
        $className = str_replace('Net_Gearman_', '', $className);
        include_once PATH_TO_PEAR_NET_GEARMAN.$className.'.php';
    }
    //Костыль для pear Net_Gearman_Job
    if(strstr($className, 'Net_Gearman_Job_')){
        $className = str_replace('Net_Gearman_Job_', '', $className);
        include_once PATH_TO_PEAR_NET_GEARMAN.'Job/'.$className.'.php';
    }

}

Это, кстати, дает возможность работать с Gearman в php, используя тот же Denwer или OpenServer.

Текущая версия — это альфа-релиз, датированный 2009 годом. Но это не страшно: никто не мешает править требуемые файлы, никакие дополнительные компоненты/библиотеки не используются.
Несмотря на дату, библиотека работоспособна на php5.3.* и доработки не требует.

Возникает вопрос: как все это счастье взаимодействует с сервером Gearman? Через сокеты, посылая серверу команды (вот описание взаимодействия с сервером на оф. сайте)

Отметим важную деталь pear Net_Gearman: в библиотеке есть средства мониторинга сервера и (частично) управления им, что дает возможность сделать, например, упомянутый выше скрипт для мониторинга.

pecl gearman

Несмотря на все достоинства, интерфейс библиотеки pear Net_Gearman беден и, как где-то выразились, несколько неуклюж. К тому же IDE — phpStorm например, классов Net_Gearman не содержит (для удобной работы надо костыль прикрутить), и для рабочего использования лучше подходит pecl gearman.
Главное отличие pecl gearman от pear Net_gearman в том, что pecl gearman с сервером напрямую не взаимодействует — это обертка для С-библиотеки libgearman.

Установка pecl gearman

Очень бы хотелость вот так вот сразу:
pecl install gearman
Но так не получится. В конце установки появится сообщение о том, что требуется libgearman версии 0.21 и выше, сборка такой библиотеки тянет за собой еще некоторые действия, и успехом все этом может не увенчаться.
В интернете много рекомендаций по установке библиотеки pecl-gearman-0.7.0, но у нее есть баг — есть большая вероятность вылететь с ошибкой Segmentation fault.
Путем нескольких проб было установлено, что устойчиво работает и без проблем устанавливается версия 0.8.1
Вот порядок установки на debian6/php5.3* «с нуля», начиная с сервера (для debian7/php5.4 см. вот эту публикацию):

aptitude install gearman-job-server
aptitude install php5-dev
aptitude install php-pear
aptitude install make
aptitude install libgearman-dev
cd /tmp
pecl download gearman-0.8.1
tar -xvf gearman-0.8.1.tgz
cd gearman-0.8.1
phpize
./configure
make
make test
make install
echo 'extension=gearman.so' > /etc/php5/conf.d/gearman.ini

Теперь можно спокойно использовать всю красоту библиотеки в реальном проекте. IDE ��одержит классы для pecl gearman, дополнительных действий не требуется. Все приведенные ниже примеры используют именно расширение pecl gearman.

Примеры

Самый простой клиент и воркер

Клиент

<?php
$client = new GearmanClient();

/*Эта ф-я вернет true независимо от того, есть такой сервер или нет. Для проверки доступности сервера нужно использовать echo(‘’), установив на всякий случай таймаут в миллисекундах во избежание затыка скрипта при недоступности сервера */
$client->addServer('192.168.68.4');


$client->setTimeout(29000);

/*true/false в зависимости от доступности сервера*/
$haveGoodServer = $client->echo('');
var_dump($haveGoodServer);

$data = ‘slon yooo’;

/*Отправляем задачу и данные на Gearman и ждем выполнения*/
$res = $client->do('function_revert_string_and_caps', $data);

/*Мы увидим результат, как только его вернет сервер, ну или выскочим по таймауту*/
echo $res;

Воркер

<?php
$worker = new GearmanWorker();
$worker->addServer('192.168.68.4');

/*Тут мы говорим, что готовы обработать ф-ю function_revert_string_and_caps, и что заниматься этим будет ф-я 'revCaps*/
$worker->addFunction('function_revert_string_and_caps', 'revCaps');

/*Запускаем воркер. В таком варианте он отработает один раз*/
$worker->work();

/*А это вариант будет висеть демоном - есть на видео*/
//while($worker->work()){};


//Ну и сама ф-я обработчик, аргумент один - объект-задание job
function revCaps($job){

   /*Извлекаем из job данные, переданные клиентом*/ 
   $content = $job->workload();
    
    return mb_strtoupper(strrev($content));
}

Видео примера

Передача данных по частям и дополнительный обмен данными с клиентом

Опе��ацией do в клиенте пользоваться не всегда удобно, зачастую лучше использовать добавление задачи и обработку данных, приходящих от сервера, с помощью простых callback ф-й.
Кроме того, если данные передаются частями, имеет смысл отобразить количество обработанных частей — например, для прогресс-бара.
Клиент

<?php
$client = new GearmanClient();
$client->addServer('192.168.68.4');

$data = '';
/*Добавляем задачу. Пока она еще не выполняется*/
$client->addTask('function_serial_send', $data);

/*Указываем, какая ф-я будет обрабатывать событие успешной постановки задачи в очередь*/
$client->setCreatedCallback('createTask');
/*Ну и сама ф-я - обработчик этого события*/
function createTask(GearmanTask $task){
    echo "Start data from ".$task->jobHandle()."\n";
}

/*Указываем, какая ф-я будет обрабатывать принимаемые данные*/
$client->setDataCallback('getData');
/*сама ф-я для обработки принимаемых данных*/
function getData(GearmanTask $task){
    echo "received:".$task->data()."\n\n";
}

/*Указываем, какая ф-я обрабатывает прием статуса Complete - то есть окончание обработки*/
$client->setCompleteCallback('stopTask');
/*Ф-я - в ней прекращаем работу скрипта*/
function stopTask(){
    echo "task Stop\n";
    exit;
}

/*Указываем, какая ф-я обрабатывает числовые данные о ходе обработки*/
$client->setStatusCallback('getStatus');
/*Результат просто показываем в %*/
function getStatus(GearmanTask $task){
    echo "Handled: ".$task->taskNumerator() / $task->taskDenominator()*(100)."%\n";
}

/*Запускаем все задачи - одну в данном случае. ВАЖНО: этот запуск должен стоять ПОСЛЕ объявлений
ф-й callback - иначе они работать не будут */
$client->runTasks();

Воркер

<?php
$worker = new GearmanWorker();
$worker->addServer('192.168.68.4');
$worker->addFunction('function_serial_send', 'sendDataAndStatus');

/*напоминаем - это цикл нужен, чтобы worker висел демоном*/
while($worker->work()){}

function sendDataAndStatus(GearmanJob $job){
    $arr = array('one','two', 'three', 'four', 'five');
    $i = 1;
    foreach ($arr as $diggo){

       /*Отправляем типа статус - будет передаваться такое 1,5   2,5 и т.д. Первый аргумент - Numerator, то бишь “числитель”, второй Denumenator - “знаменатель”. Вообще можно произвольные числа передавать, приводятся к типу long*/
        $job->sendStatus($i, count($arr));
        
       /*передаем порцию данных*/
        $job->sendData($diggo);

        echo "send data: ".$diggo, "\n";
        $i++;

        sleep(1);
    }
    /*Передаем статус завершения - прием такого статуса обрабатывается в нашем клиенте*/
    $job->sendComplete('');
}

Видео выполнения этого примера:

Фоновые задачи и приоритет

Статус задачи — обычная или фоновая (синхронная/асинхронная) и приоритет ее выполнения задаются в клиенте. Если задача добавлена на сервер как фоновая, клиент фактически просто «бросает» ее на сервер и ее дальнейшей судьбой не интересуется. Если задача — обычная, клиент ждет от сервера результата ее выполнения.
Для фоновых задач применяется приставка background.
Для приоритетов есть три уровня — обычный (без приставок), низкий/высокий cоотв. Low/High, например, метод addTaskHighBackground — добавить фоновую задачу с высоким приоритетом.
Все это хорошо расписано в оф. документации в разделе GearmanClient

Параллельные вычисления в PHP с использованием Gearman

На закуску, конечно, самое вкусное.
Код этого примера несколько объемен, посему не приводится, но ничего военного в нем нет: клиент добавляет 800+ задач на сервер очередей, 4 воркера обрабатывают задачи.
Сами задачи — перевод кусков текста с помощью Yandex Translate API, вся задача в целом — перевод книги «Крестный отец» с русского на украинский язык.
Вот видео: