УрокОбработка большого количества данных с помощью Queue API

В жизни каждого программиста попадаются заказчики, которые просят сделать для их сайта обработку большого количества данных, будь то постоянное обновление товаров для магазина, парсинг xml файлов, отправка десятков тесяч емейлов и так далее. Смысл в том, что решить задачу "в лоб" просто невозможно - сервер может не выдержать; или закончится время выполнения скрипта и обработка данных прервётся; или же (в случае рассылки емейлов) ляжет почтовый сервер, или того хуже - почтовый сервер решит, что вы спамер, и ваши письма будут доходить до адресатов с большим опозданием (если вообще дойдут).

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

В Drupal'e панацеей от этой болезни является Queue API. В седьмом Друпале оно уже находится в ядре, а для шестой версии был сделан бэкпорт в виде модуля drupal_queue.

Смысл Queue API:

• При запуске крона Queue API создаёт очередь из данных, которые надо обработать
• К сайту добавляется ещё один крон, который запускается раз в несколько минут, и обрабатывает небольшую порцию из очереди

Теперь к реализации.

Первым делом положите в корень сайта файл drupal_queue_cron.php с вот таким содержимым (его также можно взять из модуля drupal_queue):

<?php
 
/**
 * @file
 * Entry point for worker calls.
 */
include_once './includes/bootstrap.inc';
drupal_bootstrap(DRUPAL_BOOTSTRAP_FULL);
if (function_exists('drupal_queue_cron_run')) {
  drupal_queue_cron_run();
}

Настройте запуск этого файла на сервере раз в несколько минут.

Далее переходим к коду. В своем модуле надо имплементировать хук hook_cron_queue_info(), который является отправной точкой для создания очереди:

/**
 * Implements hook_cron_queue_info().
 */
function ИМЯМОДУЛЯ_cron_queue_info() {
  $queue['mymodule_queue'] = array(
    'worker callback' => 'mymodule_item_process',
    'time' => 60,
  );
  return $queue;
}

Небольшое пояснение:

• mymodule_queue - имя очереди. Просто следите за тем, чтобы оно не перекликалось с другими очередями на сайте.
• worker callback - функция, которая будет обрабатывать каждый элемент очереди.
• time - максимальное время выполнения крона очереди (дополнительного крона).

Далее имплементируем hook_cron(), который будет вызван при запуске выполнения регулярных процедур:

/**
 * Implements hook_cron().
 */
function ИМЯМОДУЛЯ_cron() {
 
  // Загружаем массив, каждый элемент которого 
  // будет являться элементом очереди
  $items = mymodule_load_data();
 
  if ($items) {
    // Создаём новую очередь
    $queue = drupal_queue_get('mymodule_queue');   
    $queue->createQueue();
 
    // Каждый элемент ставим в очередь, т.е.
    // по факту - идёт запись в базу данных
    foreach ($items as $item) {
      $queue->createItem($item);
    }
  }
}

Например, загружу емейлы всех пользователей сайта:

function mymodule_load_data() {
  // Сюда я буду ложить все данные
  $data = array();
  // Вытягиваю из бд имя и емейл каждого пользователя
  $users = db_query('SELECT name, mail FROM {users}');
  // Ложу всех 
  foreach ($users as $user) {
    $data[] = array(
      'mail' => $user->mail,
      'name' => $user->name,
    ); 
  }
  return $data;
}

Собственно, сама функция, которая будет обрабатывать элемент (worker callback):

function mymodule_item_process($data) {
  // Строю параметры отправки письма
  $site_name = variable_get('site_name', '');
  $params = array();
  $params['subject'] = t('Notification from !sitename', array('!sitename' => $site_name));
  $params['body'] = t('Hi, !username', array('!username' => $data['name'));
  // Отправляю письмо
  drupal_mail('mymodule', 'send_notifications', $data['mail'], language_default(), $params);
}

Ну и для порядка уже допишу имплементацию hook_mail(), которая должна отправлять мои письма:

/**
 * Implementats of hook_mail().
 */
function mymodule_mail($key, &$message, $params) {
  if ($key == 'send_notifications') {
    $message['subject'] = $params['subject'];
    $message['body'][] = $params['body'];
  }
}

Кстати, у меня при работе с Queue API у меня сразу возник вопрос - а сколько элементов очереди функция обрабатывает за раз? Ответ - столько, сколько успеет выполнить скрипт, пока не закончится время выполнения скрипта 'time' в хуке hook_cron_queue_info().