Drupal Queue API: Sincroniza Contenido Externo sin Bloquear tu Servidor

Drupal Queue API: procesamiento en segundo plano

Cuando trabajas con Drupal en entornos institucionales, tarde o temprano te enfrentas a un reto común: procesar grandes volúmenes de datos sin degradar la experiencia del usuario. Sincronizar contenido desde una API externa, procesar miles de registros o ejecutar migraciones complejas son operaciones que no deberían ejecutarse en el mismo hilo que una petición HTTP.

La Queue API de Drupal es la solución nativa para este tipo de escenarios. Implementa el patrón productor-consumidor: encolas tareas desde cualquier parte de tu código y Drupal las procesa de forma asíncrona durante la ejecución del cron, liberando al servidor de bloqueos y cuellos de botella.

En esta guía comparto un caso real: la integración de una API externa que dictaba qué contenido crear, actualizar y borrar en un portal institucional, gestionado íntegramente mediante la Queue API de Drupal.

Fundamentos de la Queue API

La Queue API de Drupal gira en torno a tres conceptos:

• Cola (Queue): estructura FIFO donde se almacenan los items pendientes de procesar.
• Item: cada tarea individual que se encola. Puede contener cualquier tipo de dato que PHP pueda serializar.
• Worker: la lógica que procesa cada item de la cola.

Las operaciones básicas son:

$queue = \DrupalQueue::get('nombre_de_la_cola');

// Encolar un item
$queue->createItem($datosDelItem);

// Reclamar el siguiente item pendiente
$item = $queue->claimItem();

// Eliminar el item después de procesarlo con éxito
$queue->deleteItem($item);

Drupal incluye DatabaseQueue como backend por defecto, que almacena los items en la tabla queue de la base de datos. Para entornos de alto rendimiento, puedes escalar a backends alternativos como Redis o RabbitMQ manteniendo la misma API.

Caso práctico: sincronizar contenido desde una API externa

El problema

Uno de los portales institucionales que lideré necesitaba mantenerse sincronizado con un sistema externo de gestión académica. Periódicamente, una API REST nos devolvía un array JSON con las acciones pendientes:

[
  { "action": "create", "externalId": 1401, "data": { ... } },
  { "action": "update", "externalId": 980,  "data": { ... } },
  { "action": "delete", "externalId": 423,  "data": { ... } },
  { "action": "update", "externalId": 1250, "data": { ... } }
]

La respuesta podía contener cientos de acciones en cada ciclo de sincronización. Procesarlas todas de forma síncrona durante una petición HTTP habría bloqueado el servidor durante minutos, dejando la web inaccesible.

Solución: Queue API con Worker personalizado

La estrategia fue dividir el problema en tres capas:

1. Recolector: consulta la API externa y encola cada acción como un item independiente.
2. Worker (QueueWorker): define la lógica de procesamiento para cada tipo de acción.
3. Ejecutor: cron de Drupal procesa la cola de forma gradual.

1. Definir el Worker

Creamos un plugin QueueWorker en nuestro módulo custom:

// src/Plugin/QueueWorker/ContentSyncWorker.php

namespace Drupal\mi_modulo_custom\Plugin\QueueWorker;

use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\node\Entity\Node;

/**
 * Procesa items de la cola de sincronización de contenido.
 *
 * @QueueWorker(
 *   id = "content_sync_worker",
 *   title = @Translation("Sincronización de contenido externo"),
 *   cron = {"time" = 60}
 * )
 */
class ContentSyncWorker extends QueueWorkerBase {

  public function processItem($item): void {
    switch ($item->action) {
      case 'create':
        $this->handleCreate($item);
        break;

      case 'update':
        $this->handleUpdate($item);
        break;

      case 'delete':
        $this->handleDelete($item);
        break;
    }
  }

  private function handleCreate($item): void {
    $node = Node::create([
      'type' => 'contenido_sincronizado',
      'title' => $item->data['title'],
      'field_external_id' => $item->externalId,
      'field_contenido' => $item->data['body'],
    ]);
    $node->save();
    \Drupal::logger('content_sync')->info('Contenido creado: @id', ['@id' => $item->externalId]);
  }

  private function handleUpdate($item): void {
    $nodes = \Drupal::entityTypeManager()
      ->getStorage('node')
      ->loadByProperties(['field_external_id' => $item->externalId]);

    if (!empty($nodes)) {
      $node = reset($nodes);
      $node->set('title', $item->data['title']);
      $node->set('field_contenido', $item->data['body']);
      $node->save();
      \Drupal::logger('content_sync')->info('Contenido actualizado: @id', ['@id' => $item->externalId]);
    }
  }

  private function handleDelete($item): void {
    $nodes = \Drupal::entityTypeManager()
      ->getStorage('node')
      ->loadByProperties(['field_external_id' => $item->externalId]);

    if (!empty($nodes)) {
      $node = reset($nodes);
      $node->delete();
      \Drupal::logger('content_sync')->info('Contenido eliminado: @id', ['@id' => $item->externalId]);
    }
  }

}

El worker procesa cada item de la cola durante la ejecución del cron. El parámetro cron = {"time" = 60} indica que el cron dedicará hasta 60 segundos a procesar esta cola en cada ejecución.

2. Encolar las acciones

El recolector se ejecuta mediante un hook de cron o un endpoint específico:

/**
 * Consulta la API externa y encola las acciones pendientes.
 */
function mi_modulo_custom_sync_from_external_api(): void {
  $queue = \DrupalQueue::get('content_sync_worker');

  // Consultar la API externa.
  $client = \Drupal::httpClient();
  $response = $client->get('https://api-institucional.example.com/content-sync');
  $actions = json_decode($response->getBody()->getContents());

  // Encolar cada acción recibida.
  foreach ($actions as $action) {
    $queue->createItem((object) [
      'action' => $action->action,
      'externalId' => $action->externalId,
      'data' => $action->data,
    ]);
  }

  $count = count($actions);
  \Drupal::logger('content_sync')->info('Encoladas @count acciones de sincronización.', ['@count' => $count]);
}

3. Monitorear y procesar

La cola se procesa automáticamente en cada ejecución del cron de Drupal. Para ejecutar el cron manualmente desde consola:

drush cron

También puedes forzar el procesamiento de una cola concreta si necesitas un control más fino:

drush queue:run content_sync_worker

El parámetro --time-limit limita el tiempo de procesamiento en segundos:

drush queue:run content_sync_worker --time-limit=5

Gestionar colas desde drush

Además de ejecutar colas, drush ofrece varios comandos para monitorizar y administrar colas en producción:

Listar todas las colas del sistema:

drush queue:list

Muestra el nombre de cada cola, el número de items pendientes y la clase que la implementa:

--------------------- ------- ---------------------------------
Queue                   Items  Class
--------------------- ------- ---------------------------------
content_sync_worker      125  Drupal\Core\Queue\DatabaseQueue
locale_translation         0  Drupal\Core\Queue\DatabaseQueue
--------------------- ------- ---------------------------------

Eliminar todos los items de una cola:

drush queue:delete content_sync_worker

Esto vacía completamente la cola. Útil cuando necesitas reiniciar una sincronización desde cero o limpiar items huérfanos tras un cambio en la lógica de procesamiento.

Generar un plugin QueueWorker desde drush:

drush generate plugin-queue-worker

El generador interactivo te preguntará el nombre del módulo, la etiqueta y el ID del plugin, y creará automáticamente el esqueleto de la clase QueueWorker con el método processItem() listo para que añadas tu lógica.

Módulo Queue UI

Aunque drush cubre la mayoría de necesidades, el módulo Queue UI (drush en queue_ui) añade una interfaz gráfica en la administración de Drupal (/admin/config/system/queue-ui) desde donde puedes:

• Ver el número de items en cada cola.
• Inspeccionar el contenido de cada item.
• Procesar colas manualmente desde la interfaz.

Para entornos de producción donde varios miembros del equipo necesitan visibilidad sin acceso a consola, Queue UI es una herramienta muy recomendable.

Gestión de errores y robustez

En un escenario de sincronización con una API externa, los fallos son inevitables: la API puede no responder, un registro puede estar corrupto o la conexión de red puede interrumpirse.

Estrategias de tolerancia a fallos que aplicamos en producción:

1. Validación temprana

Antes de encolar, validamos que los datos tengan la estructura esperada. Un item mal formado no debería llegar nunca a la cola.

foreach ($actions as $action) {
  // Validar campos obligatorios antes de encolar.
  if (empty($action->action) || empty($action->externalId)) {
    \Drupal::logger('content_sync')->warning('Acción descartada: datos incompletos.');
    continue;
  }

  $queue->createItem((object) [
    'action' => $action->action,
    'externalId' => $action->externalId,
    'data' => $action->data ?? [],
  ]);
}

2. Manejo de excepciones en el Worker

Si un item lanza una excepción, Drupal lo marca como no procesado y lo reintenta en la siguiente ejecución del cron. Por eso es crítico que el worker sea idempotente: procesar el mismo item varias veces no debe producir efectos secundarios no deseados.

public function processItem($item): void {
  try {
    switch ($item->action) {
      // ...
    }
  }
  catch (\Exception $e) {
    \Drupal::logger('content_sync')->error(
      'Error al procesar item @id: @message',
      ['@id' => $item->externalId ?? 'desconocido', '@message' => $e->getMessage()]
    );
    // Relanzar para que Drupal no marque el item como procesado.
    throw $e;
  }
}

3. Idempotencia

La lógica de create usa field_external_id para evitar duplicados si el item se procesa dos veces:

private function handleCreate($item): void {
  // Verificar si ya existe antes de crear.
  $existing = \Drupal::entityTypeManager()
    ->getStorage('node')
    ->loadByProperties(['field_external_id' => $item->externalId]);

  if (!empty($existing)) {
    \Drupal::logger('content_sync')->info('Contenido @id ya existe, se omite la creación.', ['@id' => $item->externalId]);
    return;
  }

  // Crear el nodo normalmente.
  // ...
}

4. Monitorización y alertas

La Queue API expone el tamaño de la cola, lo que permite activar alertas si se acumulan items sin procesar:

$queue = \DrupalQueue::get('content_sync_worker');
$pendingItems = $queue->numberOfItems();

if ($pendingItems > 500) {
  // Enviar alerta al equipo de operaciones.
  \Drupal::logger('content_sync')->alert('@count items pendientes en la cola de sincronización.', ['@count' => $pendingItems]);
}

Rendimiento y escalabilidad

La DatabaseQueue por defecto es adecuada para la mayoría de escenarios institucionales. Sin embargo, cuando el volumen de items supera los cientos de miles diarios, conviene migrar a backends especializados:

Backend	Ventajas	Cuándo usarlo
DatabaseQueue (core)	Sin dependencias externas, simple	Volúmenes moderados (< 10.000 items/día)
Redis	En memoria, muy rápido	Alta frecuencia de encolado/procesamiento
RabbitMQ	Distribuido, persistente	Arquitecturas multi-servidor, prioridades

La migración entre backends es transparente para tu código: la Queue API abstrae el motor de almacenamiento detrás de la misma interfaz.

¿Puedo integrar una API externa sin colas?

Sí, y en muchos casos es correcto hacerlo. La Queue API es necesaria cuando:

• El volumen de datos a procesar es alto (cientos o miles de registros por ciclo).
• La API externa tiene latencias variables o puede fallar temporalmente.
• Quieres desacoplar la sincronización de la petición HTTP del usuario.
• Necesitas trazabilidad y reintentos automáticos.

En cambio, puedes prescindir de colas si:

• Sincronizas pocos registros (menos de 10) y la operación es rápida.
• La sincronización forma parte de una acción del usuario (por ejemplo, tras pulsar un botón).
• No necesitas reintentos ni trazabilidad detallada.

Lecciones aprendidas

Tras implementar este patrón en producción, estas son las conclusiones que me llevo:

• Valida siempre antes de encolar. Un item mal formado en la cola se convierte en un fallo silencioso difícil de depurar.
• Diseña para la idempotencia. El cron puede interrumpirse a mitad de procesamiento, y Drupal reintenta los items. Tu worker debe tolerar ser ejecutado varias veces sobre el mismo item.
• Usa logging estructurado. Drupal::logger() te permite filtrar por canal y gravedad. En producción, vale su peso en oro.
• Rompe el trabajo en items pequeños. No encoles un array de 100 acciones como un solo item. Cada acción debería ser un item independiente para que el cron pueda procesarlos gradualmente y los reintentos sean granulares.
• Monitoriza el tamaño de la cola. Una cola que crece sin control es un síntoma temprano de un problema. Activa alertas automáticas.

---

La Queue API es una de esas herramientas de Drupal que marca la diferencia cuando trabajas con integraciones complejas y procesamiento de datos a escala. Si tu portal depende de contenido externo o necesitas ejecutar operaciones pesadas en segundo plano, dominarla te ahorrará dolores de cabeza en producción.

Si necesitas importar contenido desde un sistema externo antes de procesarlo con colas, la guía de Migrate API cubre cómo hacerlo paso a paso con JSON.