Queues are a real help whenever we need to set some work aside and come back to it later. The idea is simple: we push tasks or data into a queue (we create the queue), then process them afterwards with a QueueWorker plugin (we drain the queue), usually on cron. In this post we'll see how to do both.
There are several ways to fill a queue:
- from a form,
- from a controller,
- from a
hook_cron()implementation.
And several ways to process it:
- on cron with a QueueWorker plugin,
- as a Batch process with a QueueWorker plugin,
- by claiming each item manually inside a service or controller.
Here we'll fill the queue from a controller and drain it with a QueueWorker plugin on cron. Why from a controller? Because you can then trigger it from an external scheduler, a Linux crontab for instance, instead of leaning on Drupal's "poor man's cron", which only fires on page requests.
The module imports the title and description of each entry in the Drupal Planet RSS feed, pushes them onto a queue named exqueue_import, and, when cron runs, creates one page node per item. The original Drupal 8 example lives at github.com/KarimBoudjema/Drupal8-ex-queue-api-01; the code below is rewritten for Drupal 11.
The module has two moving parts:
- a controller
src/Controller/ExQueueController.phpwith a route inexqueue01.routing.ymland two methods:getData()(fetch data and enqueue it) anddeleteTheQueue()(empty the queue); - a QueueWorker plugin
src/Plugin/QueueWorker/ExQueue01.phpthat processes each item.
web/modules/custom/exqueue01/
|-- exqueue01.info.yml
|-- exqueue01.routing.yml
`-- src
|-- Controller
| `-- ExQueueController.php
`-- Plugin
`-- QueueWorker
`-- ExQueue01.phpFirst, let's scaffold the module and its plugins with Drush 13's code generator (Drupal Console is long gone):
drush generate module
drush generate controller
drush generate plugin:queue-worker1. Fill the queue from a controller
Here is the Drupal 11 controller. Note the modern dependency injection: the class uses the core AutowireTrait, so we no longer hand-write a create() method: the #[Autowire] attributes on the promoted constructor properties tell the container which services to pass in.
<?php
declare(strict_types=1);
namespace Drupal\exqueue01\Controller;
use Drupal\Core\Controller\ControllerBase;
use Drupal\Core\DependencyInjection\AutowireTrait;
use Drupal\Core\Queue\QueueFactory;
use GuzzleHttp\ClientInterface;
use GuzzleHttp\Exception\RequestException;
use Symfony\Component\DependencyInjection\Attribute\Autowire;
/**
* Demonstrates the Queue API.
*
* getData() loads external data and creates one queue item per entry in the
* "exqueue_import" queue. deleteTheQueue() empties that queue. On cron run the
* ExQueue01 queue worker turns each item into a page node.
*/
final class ExQueueController extends ControllerBase {
use AutowireTrait;
public function __construct(
#[Autowire(service: 'queue')]
protected readonly QueueFactory $queueFactory,
#[Autowire(service: 'http_client')]
protected readonly ClientInterface $httpClient,
) {}
/**
* Deletes the "exqueue_import" queue and all of its items.
*/
public function deleteTheQueue(): array {
$this->queueFactory->get('exqueue_import')->deleteQueue();
return ['#markup' => $this->t('The queue "exqueue_import" has been deleted.')];
}
/**
* Loads external data and pushes one queue item per entry.
*/
public function getData(): array {
// 1. Get the data as an array of objects (swap in getFakeData() locally).
$data = $this->getDataFromRss();
if (!$data) {
return ['#markup' => $this->t('No data found.')];
}
// 2. Get the queue and count its items before we add anything.
$queue = $this->queueFactory->get('exqueue_import');
$totalBefore = $queue->numberOfItems();
// 3. Create one queue item per entry.
foreach ($data as $element) {
$queue->createItem($element);
}
$totalAfter = $queue->numberOfItems();
// 4. Show what is now in the queue.
$list = $this->getItemList($queue);
return [
'#type' => 'table',
'#caption' => $this->t('The queue had @before item(s). We added @count. It now holds @after.', [
'@before' => $totalBefore,
'@count' => count($data),
'@after' => $totalAfter,
]),
'#header' => [$this->t('Title'), $this->t('ID')],
'#rows' => $list,
'#empty' => $this->t('No items.'),
'#sticky' => TRUE,
];
}
/**
* Builds a fake data set, handy for local testing without network access.
*/
protected function getFakeData(): array {
$content = [];
for ($i = 1; $i <= 10; $i++) {
$item = new \stdClass();
$item->title = 'Title ' . $i;
$item->body = 'Body ' . $i;
$content[] = $item;
}
return $content;
}
/**
* Fetches the Drupal Planet RSS feed and returns an array of item objects.
*/
protected function getDataFromRss(): array {
$uri = 'https://www.drupal.org/planet/rss.xml';
try {
$response = $this->httpClient->get($uri, ['headers' => ['Accept' => 'text/plain']]);
$body = (string) $response->getBody();
}
catch (RequestException) {
return [];
}
if ($body === '') {
return [];
}
$xml = simplexml_load_string($body);
if ($xml === FALSE) {
return [];
}
$content = [];
foreach ($xml->children()->children() as $child) {
if (!empty($child->title)) {
$item = new \stdClass();
$item->title = (string) $child->title;
$item->body = (string) $child->description;
$content[] = $item;
}
}
return $content;
}
/**
* Claims every item to display it, then releases the claims.
*/
protected function getItemList($queue): array {
$rows = [];
$claimed = [];
// claimItem() also leases (locks) the item for one hour by default, so we
// must release each claim afterwards.
while ($item = $queue->claimItem()) {
$rows[] = [$item->data->title, $item->item_id];
$claimed[] = $item;
}
foreach ($claimed as $item) {
$queue->releaseItem($item);
}
return $rows;
}
}We inject two services here: QueueFactory to work with the queue, and Guzzle's http_client to fetch the RSS feed. Messages are printed with $this->messenger(), which ControllerBase hands us for free.
Here comes the interesting part, and it all happens in getData():
$queue = $this->queueFactory->get('exqueue_import');
$totalBefore = $queue->numberOfItems();
foreach ($data as $element) {
$queue->createItem($element);
}Three lines do the real work. $this->queueFactory->get('exqueue_import') returns the default QueueInterface backend for a queue with that name, creating it on first use. numberOfItems() tells us how many items are waiting, and createItem() pushes one item onto the queue.
To show the queue we then read it back, claiming each item and releasing it right away so it stays available for the real worker:
while ($item = $queue->claimItem()) {
$rows[] = [$item->data->title, $item->item_id];
$claimed[] = $item;
}
foreach ($claimed as $item) {
$queue->releaseItem($item);
}Visit /exqueue01/getData and the queue fills up. That's not too difficult, is it?
To inspect the queue from the CLI, Drush 13 ships a queue toolkit: drush queue:list shows every registered queue and its item count.
2. Drain the queue with a QueueWorker plugin
So far so good, but a queue nobody empties is not much use. Now we need a QueueWorker to process each item on cron. In Drupal 11 the plugin is declared with the #[QueueWorker] PHP attribute: annotations were removed in Drupal 10, so the old @QueueWorker doc-block no longer works.
<?php
declare(strict_types=1);
namespace Drupal\exqueue01\Plugin\QueueWorker;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\Queue\Attribute\QueueWorker;
use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Symfony\Component\DependencyInjection\ContainerInterface;
/**
* Turns each queued RSS item into a page node.
*/
#[QueueWorker(
id: 'exqueue_import',
title: new TranslatableMarkup('Import Content From RSS'),
cron: ['time' => 5],
)]
final class ExQueue01 extends QueueWorkerBase implements ContainerFactoryPluginInterface {
public function __construct(
array $configuration,
$plugin_id,
$plugin_definition,
protected readonly EntityTypeManagerInterface $entityTypeManager,
protected readonly LoggerChannelFactoryInterface $loggerFactory,
) {
parent::__construct($configuration, $plugin_id, $plugin_definition);
}
public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition): self {
return new self(
$configuration,
$plugin_id,
$plugin_definition,
$container->get('entity_type.manager'),
$container->get('logger.factory'),
);
}
public function processItem($data): void {
$title = $data->title ?? NULL;
$body = $data->body ?? NULL;
// If the payload is unusable, throwing keeps the item in the queue; here we
// deliberately swallow the error so the useless item is dropped instead.
try {
if (!$title || !$body) {
throw new \InvalidArgumentException('Missing title or body.');
}
$node = $this->entityTypeManager->getStorage('node')->create([
'type' => 'page',
'title' => $title,
'body' => ['value' => $body, 'format' => 'basic_html'],
]);
$node->save();
$this->loggerFactory->get('exqueue01')->info('Created node @id from a queue item.', [
'@id' => $node->id(),
]);
}
catch (\Exception $e) {
$this->loggerFactory->get('exqueue01')->warning('Skipped a queue item: @error', [
'@error' => $e->getMessage(),
]);
}
}
}The attribute wires the plugin to the queue:
#[QueueWorker(
id: 'exqueue_import',
title: new TranslatableMarkup('Import Content From RSS'),
cron: ['time' => 5],
)]id is the machine name of the queue this worker drains. The cron key tells Drupal to run the worker on cron and allocates up to 5 seconds per run; any items left over are picked up on the next cron pass.
A note on exceptions. On cron, core's Cron::processQueues() claims each item and calls your processItem() inside a try/catch. What you throw changes the outcome:
- let an exception bubble up and the item is kept in the queue and logged: good when a failure is transient;
- throw
\Drupal\Core\Queue\RequeueExceptionto immediately requeue the item; - throw
\Drupal\Core\Queue\SuspendQueueExceptionto stop processing the whole queue (e.g. a remote API is down); - throw
\Drupal\Core\Queue\DelayedRequeueExceptionto keep the lease but retry later.
In our worker we catch the exception ourselves, so a bad item is simply dropped rather than left in the queue forever. That is a design choice: an item with no title or body is worthless to us.
3. Run the queue
Two ways to drain it:
- Cron: because we set the
cronkey, the worker runs automatically on every cron pass (drush cron, or the scheduled cron). - Drush 13: run it on demand with
drush queue:run exqueue_import. This replaces the olddrupal queue:runDrupal Console command.
Recap. We built a controller that fetches the Drupal Planet RSS feed, created a queue with $this->queueFactory->get('exqueue_import'), pushed an item per entry with createItem(), and wrote a cron QueueWorker plugin whose processItem() turns each item into a page node. Modernised for Drupal 11, the only structural changes from the Drupal 8 original are the #[QueueWorker] attribute, autowired dependency injection, and the Drush 13 CLI.
Have you filled a queue another way, from a form or straight from a hook_cron() implementation? I'd love to hear about it in the comments.