The Drupal Queue API: Deferring Work Safely with Claim and Deletefor Drupal 11 , and 10

Every module you've written so far has done its work immediately: a page loads, your code runs, a response goes back to the browser. But what about work that's too slow, too unpredictable, or too unimportant to make a visitor wait for? Sending a batch of emails. Pinging a slow external API. Processing an uploaded file. This lesson introduces the tool Drupal gives you for exactly that situation: the Queue API.

What you'll learn in this lesson

• What a queue actually is, and why "defer it" is often the right answer to a slow operation
• How to get a queue object and add items to it with createItem()
• The claim-then-delete pattern for safely processing queue items — and why it protects you from crashes mid-task
• How Drupal automatically retries an item if something goes wrong before you finish with it

The problem a queue solves

Imagine a "welcome email" feature: every time someone registers, your module sends them an email. Sending email is slow and can fail — a mail server might be down, or just sluggish. If you send it directly inside the registration request, the visitor sits there staring at a spinner waiting for a mail server they've never heard of.

A queue lets you flip that around. Instead of doing the slow work immediately, you drop a small note onto a list — "send a welcome email to this address" — and return control to the visitor instantly. Something else (usually Drupal's cron system, which you'll meet properly in a later lesson) comes along afterward, works through that list one item at a time, and does the actual slow work in the background. The visitor never waits for it.

The source file

Path (relative to the Examples module's root): modules/queue_example/src/Form/QueueExampleForm.php

<?php

namespace Drupal\queue_example\Form;

use Drupal\Component\Utility\Html;
use Drupal\Core\CronInterface;
use Drupal\Core\Database\Connection;
use Drupal\Core\Form\FormBase;
use Drupal\Core\Form\FormStateInterface;
use Drupal\Core\Queue\DatabaseQueue;
use Drupal\Core\Queue\QueueFactory;
use Drupal\Core\Queue\QueueGarbageCollectionInterface;
use Drupal\Core\Site\Settings;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Form with examples on how to use queue.
 */
class QueueExampleForm extends FormBase {

  /**
   * The queue factory.
   *
   * @var \Drupal\Core\Queue\QueueFactory
   */
  protected $queueFactory;

  /**
   * The database connection.
   *
   * @var \Drupal\Core\Database\Connection
   */
  protected $database;

  /**
   * The cron service.
   *
   * @var \Drupal\Core\CronInterface
   */
  protected $cron;

  /**
   * The type of queue backend.
   *
   * @var string
   */
  protected $queueType;

  /**
   * Constructs a new \Drupal\queue_example\Form\QueueExampleForm object.
   *
   * @param \Drupal\Core\Queue\QueueFactory $queue_factory
   *   The queue factory.
   * @param \Drupal\Core\Database\Connection $database
   *   The database connection.
   * @param \Drupal\Core\CronInterface $cron
   *   The cron service.
   * @param \Drupal\Core\Site\Settings $settings
   *   The site settings.
   */
  public function __construct(
    QueueFactory $queue_factory,
    Connection $database,
    CronInterface $cron,
    Settings $settings,
  ) {
    $this->queueFactory = $queue_factory;
    $this->queueType = $settings->get('queue_default', 'queue.database');
    $this->database = $database;
    $this->cron = $cron;
  }

  /**
   * {@inheritdoc}
   */
  public static function create(ContainerInterface $container) {
    $form = new static(
      $container->get('queue'),
      $container->get('database'),
      $container->get('cron'),
      $container->get('settings')
    );
    $form->setMessenger($container->get('messenger'));

    return $form;
  }

  /**
   * {@inheritdoc}
   */
  public function getFormId() {
    // Return a string that is the unique ID of our form. Best practice here is
    // to namespace the form based on your module's name.
    return 'queue_example';
  }

  /**
   * {@inheritdoc}
   */
  public function buildForm(array $form, FormStateInterface $form_state) {
    $queue_name = $form_state->getValue('queue_name') ?: 'queue_example_first_queue';
    $items = $this->retrieveQueue($queue_name);

    $form['help'] = [
      '#type' => 'markup',
      '#markup' => '<div>' . $this->t(
        'This page is an interface on the Drupal queue API. You can add new items to the queue, "claim" one (retrieve the next item and keep a lock on it), and delete one (remove it from the queue). Note that claims are not expired until cron runs, so there is a special button to run cron to perform any necessary expirations.'
      ) . '</div>',
    ];

    $form['wrong_queue_warning'] = [
      '#type' => 'markup',
      '#markup' => '<div>' . $this->t(
        'Note: the example works only with the default queue implementation, which is not currently configured.'
      ) . '</div>',
      '#access' => (!$this->doesQueueUseDB()),
    ];

    $queue_names = ['queue_example_first_queue', 'queue_example_second_queue'];
    $form['queue_name'] = [
      '#type' => 'select',
      '#title' => $this->t('Choose queue to examine'),
      '#options' => array_combine($queue_names, $queue_names),
      '#default_value' => $queue_name,
    ];

    $form['queue_show'] = [
      '#type' => 'submit',
      '#value' => $this->t('Show queue'),
      '#submit' => ['::submitShowQueue'],
    ];

    $form['status_fieldset'] = [
      '#type' => 'fieldset',
      '#title' => $this->t('Queue status for @name', ['@name' => $queue_name]),
      '#collapsible' => TRUE,
    ];

    if (count($items) > 0) {
      $form['status_fieldset']['status'] = [
        '#theme' => 'table',
        '#header' => [
          $this->t('Item ID'),
          $this->t('Claimed/Expiration'),
          $this->t('Created'),
          $this->t('Content/Data'),
        ],
        '#rows' => array_map([$this, 'processQueueItemForTable'], $items),
      ];
    }
    else {
      $form['status_fieldset']['status'] = [
        '#type' => 'markup',
        '#markup' => $this->t('There are no items in the queue.'),
      ];
    }

    $form['insert_fieldset'] = [
      '#type' => 'fieldset',
      '#title' => $this->t('Insert into @name', ['@name' => $queue_name]),
    ];

    $form['insert_fieldset']['string_to_add'] = [
      '#type' => 'textfield',
      '#size' => 10,
      '#default_value' => $this->t('Queue item'),
    ];

    $form['insert_fieldset']['add_item'] = [
      '#type' => 'submit',
      '#value' => $this->t('Insert into queue'),
      '#submit' => ['::submitAddQueueItem'],
    ];

    $form['claim_fieldset'] = [
      '#type' => 'fieldset',
      '#title' => $this->t('Claim from queue'),
      '#collapsible' => TRUE,
    ];

    $form['claim_fieldset']['claim_time'] = [
      '#type' => 'radios',
      '#title' => $this->t('Claim time, in seconds'),
      '#options' => [
        0 => $this->t('none'),
        5 => $this->t('5 seconds'),
        60 => $this->t('60 seconds'),
      ],
      '#description' => $this->t(
        'This time is only valid if cron runs during this time period. You can run cron manually below.'
      ),
      '#default_value' => $form_state->getValue('claim_time') ?: 5,
    ];

    $form['claim_fieldset']['claim_item'] = [
      '#type' => 'submit',
      '#value' => $this->t('Claim the next item from the queue'),
      '#submit' => ['::submitClaimItem'],
    ];

    $form['claim_fieldset']['claim_and_delete_item'] = [
      '#type' => 'submit',
      '#value' => $this->t('Claim the next item and delete it'),
      '#submit' => ['::submitClaimDeleteItem'],
    ];

    $form['claim_fieldset']['run_cron'] = [
      '#type' => 'submit',
      '#value' => $this->t('Run cron manually to expire claims'),
      '#submit' => ['::submitRunCron'],
    ];

    $form['delete_queue'] = [
      '#type' => 'submit',
      '#value' => $this->t('Delete the queue and items in it'),
      '#submit' => ['::submitDeleteQueue'],
    ];

    return $form;
  }

  /**
   * Retrieves the queue from the database for display purposes only.
   *
   * It is not recommended to access the database directly, and this is only
   * here so that the user interface can give a good idea of what's going on
   * in the queue.
   *
   * @param string $queue_name
   *   The name of the queue from which to fetch items.
   *
   * @return array
   *   An array of item arrays.
   */
  public function retrieveQueue($queue_name) {
    $items = [];

    // This example requires the default queue implementation to work,
    // so we bail if some other queue implementation has been installed.
    if (!$this->doesQueueUseDb()) {
      return $items;
    }

    // Make sure there are queue items available. The queue will not create our
    // database table if there are no items.
    if ($this->queueFactory->get($queue_name)->numberOfItems() >= 1) {
      $result = $this->database->query(
        'SELECT item_id, data, expire, created FROM {' . DatabaseQueue::TABLE_NAME . '} WHERE name = :name ORDER BY item_id',
        [':name' => $queue_name],
        ['fetch' => \PDO::FETCH_ASSOC]
      );
      foreach ($result as $item) {
        $items[] = $item;
      }
    }

    return $items;
  }

  /**
   * Verifies we are using the default database queue.
   *
   * @return bool
   *   TRUE if we are using the default database queue implementation, FALSE
   *   otherwise.
   */
  protected function doesQueueUseDb() {
    return $this->queueType == 'queue.database';
  }

  /**
   * {@inheritdoc}
   */
  public function submitForm(array &$form, FormStateInterface $form_state) {
    // This method is intentionally empty, as it is not used.
    // It must be implemented because it is required by
    // \Drupal\Core\Form\FormInterface, but the parent class does not implement
    // it, contrary to validateForm(), which is implemented by the parent class
    // (\Drupal\Core\Form\FormBase).
  }

  /**
   * Submission handler for the show-queue button.
   *
   * @param array $form
   *   The form definition array.
   * @param \Drupal\Core\Form\FormStateInterface $form_state
   *   The form state object.
   */
  public function submitShowQueue(array &$form, FormStateInterface $form_state) {
    $queue = $this->queueFactory->get($form_state->getValue('queue_name'));
    // There is no harm in trying to recreate existing.
    $queue->createQueue();
    $form_state->setRebuild();
  }

  /**
   * Submission handler for the insert-into-queue button.
   *
   * @param array $form
   *   Form definition array.
   * @param \Drupal\Core\Form\FormStateInterface $form_state
   *   Form state object.
   */
  public function submitAddQueueItem(array &$form, FormStateInterface $form_state) {
    // Get a queue (of the default type) called 'queue_example_queue'.
    // If the default queue class is SystemQueue, this creates a queue that
    // stores its items in the database.
    $queue = $this->queueFactory->get($form_state->getValue('queue_name'));
    // There is no harm in trying to recreate existing.
    $queue->createQueue();

    // Queue the string.
    $queue->createItem($form_state->getValue('string_to_add'));
    $count = $queue->numberOfItems();

    $this->messenger()->addMessage(
      $this->t(
        'Queued your string (@string_to_add). There are now @count items in the queue.',
        ['@count' => $count, '@string_to_add' => $form_state->getValue('string_to_add')]
      )
    );
    // Allows us to keep information in $form_state.
    $form_state->setRebuild();
  }

  /**
   * Submission handler for the "claim" button.
   *
   * Claims (retrieves) an item from the queue and reports the results.
   *
   * @param array $form
   *   Form definition array.
   * @param \Drupal\Core\Form\FormStateInterface $form_state
   *   Form state object.
   */
  public function submitClaimItem(array &$form, FormStateInterface $form_state) {
    $queue = $this->queueFactory->get($form_state->getValue('queue_name'));
    // There is no harm in trying to recreate existing.
    $queue->createQueue();
    $item = $queue->claimItem($form_state->getValue('claim_time'));
    $count = $queue->numberOfItems();

    if (!empty($item)) {
      $this->messenger()->addMessage(
        $this->t(
          'Claimed item id=@item_id string=@string for @seconds seconds. There are @count items in the queue.', [
            '@count' => $count,
            '@item_id' => $item->item_id,
            '@string' => $item->data,
            '@seconds' => $form_state->getValue('claim_time'),
          ]
        )
      );
    }
    else {
      $this->messenger()->addMessage(
        $this->t(
          'There were no items in the queue available to claim. There are @count items in the queue.',
          ['@count' => $count]
        )
      );
    }

    $form_state->setRebuild();
  }

  /**
   * Submit function for "Claim and delete" button.
   *
   * @param array $form
   *   Form definition array.
   * @param \Drupal\Core\Form\FormStateInterface $form_state
   *   Form state object.
   */
  public function submitClaimDeleteItem(array &$form, FormStateInterface $form_state) {
    $queue = $this->queueFactory->get($form_state->getValue('queue_name'));
    // There is no harm in trying to recreate existing.
    $queue->createQueue();
    $count = $queue->numberOfItems();
    $item = $queue->claimItem(60);

    if (!empty($item)) {
      $this->messenger()->addMessage(
        $this->t(
          'Claimed and deleted item id=@item_id string=@string for @seconds seconds. There are @count items in the queue.', [
            '@count' => $count,
            '@item_id' => $item->item_id,
            '@string' => $item->data,
            '@seconds' => $form_state->getValue('claim_time'),
            ]
        )
      );
      $queue->deleteItem($item);
      $count = $queue->numberOfItems();
      $this->messenger()->addMessage(
        $this->t(
          'There are now @count items in the queue.',
          ['@count' => $count]
        )
      );
    }
    else {
      $count = $queue->numberOfItems();
      $this->messenger()->addMessage(
        $this->t(
          'There were no items in the queue available to claim/delete. There are currently @count items in the queue.',
          ['@count' => $count]
        )
      );
    }

    $form_state->setRebuild();
  }

  /**
   * Submission handler for "run cron" button.
   *
   * Runs cron (to release expired claims) and reports the results.
   *
   * @param array $form
   *   Form definition array.
   * @param \Drupal\Core\Form\FormStateInterface $form_state
   *   Form state object.
   */
  public function submitRunCron(array &$form, FormStateInterface $form_state) {
    $this->cron->run();
    $queue = $this->queueFactory->get($form_state->getValue('queue_name'));

    // @see https://www.drupal.org/node/2705809
    if ($queue instanceof QueueGarbageCollectionInterface) {
      $queue->garbageCollection();
    }
    // There is no harm in trying to recreate existing.
    $queue->createQueue();
    $count = $queue->numberOfItems();
    $this->messenger()->addMessage(
      $this->t(
        'Ran cron. If claimed items expired, they should be expired now. There are now @count items in the queue',
        ['@count' => $count]
      )
    );
    $form_state->setRebuild();
  }

  /**
   * Submission handler for clearing/deleting the queue.
   *
   * @param array $form
   *   Form definition array.
   * @param \Drupal\Core\Form\FormStateInterface $form_state
   *   Form state object.
   */
  public function submitDeleteQueue(array &$form, FormStateInterface $form_state) {
    $queue = $this->queueFactory->get($form_state->getValue('queue_name'));
    $queue->deleteQueue();
    $this->messenger()->addMessage(
      $this->t(
        'Deleted the @queue_name queue and all items in it',
        ['@queue_name' => $form_state->getValue('queue_name')]
      )
    );
  }

  /**
   * Helper method to format a queue item for display in a summary table.
   *
   * @param array $item
   *   Queue item array with keys for item_id, expire, created, and data.
   *
   * @return array
   *   An array with the queue properties in the right order for display in a
   *   summary table.
   */
  private function processQueueItemForTable(array $item) {
    if ($item['expire'] > 0) {
      $item['expire'] = $this->t(
        'Claimed: expires %expire',
        ['%expire' => date('r', $item['expire'])]
      );
    }
    else {
      $item['expire'] = $this->t('Unclaimed');
    }

    $item['created'] = date('r', $item['created']);
    $item['content'] = Html::escape(unserialize($item['data']));
    unset($item['data']);

    return $item;
  }

}

How it works

The four services this form needs

The constructor asks the service container for four things: $queueFactory (the actual API you'll use — everything routes through it), $database (used only so this demo page can peek at the raw database table for display purposes — you would never do this in real code), $cron (so a button on this page can trigger a cron run on demand), and $settings (to check which queue backend is active — the demo only works with the default database-backed queue).

Getting a queue and creating an item

$queue = $this->queueFactory->get($form_state->getValue('queue_name'));
$queue->createQueue();
$queue->createItem($form_state->getValue('string_to_add'));
$count = $queue->numberOfItems();

$queueFactory->get('some_name') returns a queue object for that name — queues don't need to be declared anywhere in advance, you just start using a name and Drupal creates the underlying storage on demand. createQueue() is safe to call every single time, even if the queue already exists, which is why you'll see it called defensively before almost every operation in this file. createItem($data) accepts any serializable PHP value (a string, an array, an object) and drops it onto the end of the queue — Drupal serializes it for you automatically.

Claiming an item — the safe way to process work

$item = $queue->claimItem($form_state->getValue('claim_time'));

claimItem($lease_time) is the read operation, and it's more careful than it looks. It atomically grabs the next unclaimed item from the front of the queue and puts a temporary lock on it for $lease_time seconds — during that window, no other process can claim the same item. This matters the moment you have more than one worker (multiple cron runs, multiple servers) potentially pulling from the same queue at once: without a lock, two workers could grab and process the same item twice.

If nothing is available to claim, claimItem() returns FALSE — always check for that before touching $item->item_id or $item->data, or you'll trigger a PHP error on an empty queue.

Claim, then delete: the real processing pattern

$item = $queue->claimItem(60);
if (!empty($item)) {
    // ... process the item here ...
    $queue->deleteItem($item);
}

This two-step dance — claim first, delete only after you're done — is the entire point of the queue's resilience. If your server crashes, times out, or throws an exception between claimItem() and deleteItem(), the item is never permanently lost. It simply sits there locked until the lease time you passed to claimItem() expires, at which point it automatically becomes claimable again for the next worker to try. You get "at least once" processing for free, without writing a single line of retry logic yourself.

Expiring stale claims with cron

$this->cron->run();
if ($queue instanceof QueueGarbageCollectionInterface) {
    $queue->garbageCollection();
}

Claim expiry doesn't happen instantly and automatically the moment a lease time passes — something has to actively go check for and release expired claims, and that something is cron. This example wires a "Run cron manually" button directly to $this->cron->run() so you can trigger that expiry check on demand, purely for the sake of this demo. In production, this happens automatically whenever cron runs on its normal schedule (a topic for the next lesson).

See it for yourself

Visit /examples/queue_example on your DDEV site, type something into the "Insert into" box, and click "Insert into queue" a couple of times.

That's a real queue with two real items sitting in it — "Process order #1042" and "Send welcome email to new-user@example.com" — exactly the kind of deferred-work notes this lesson has been describing. Notice the "Claimed/Expiration" column reads "Unclaimed" for both: nobody has called claimItem() on them yet. Try clicking "Claim the next item from the queue" and watch a status message report which item got claimed, and then "Claim the next item and delete it" to see one disappear from the table permanently.

Quick check: if your server crashes halfway through processing a claimed item — after claimItem() but before deleteItem() — is that item lost forever? If you said no, it becomes claimable again once the lease expires, you've got the core idea of this lesson.

Key takeaways

• Get a queue with $queueFactory->get('queue_name') and call createQueue() before any operation — it's idempotent, so calling it defensively every time is the normal pattern.
• createItem($data) enqueues any serializable PHP value; Drupal handles serialization for you automatically.
• Always use the claim-then-delete pattern for processing: claim it, do the work, then delete it. If processing fails before you delete, the item is automatically re-queued once its lease expires.
• claimItem() returns FALSE when nothing is available — check for that before accessing item properties.
• Claim expiry is only enforced when something actively checks for it — normally a routine cron run, which you'll see properly in the next lesson.
• A single Drupal form can wire multiple submit buttons to different handler methods via the '#submit' key, letting one page host several distinct actions cleanly.

Coming up next

A queue is only useful once something actually works through it. Next up: the Batch API — Drupal's tool for processing large amounts of work across multiple page requests with a visible progress bar, so neither your server nor your users' patience runs out.