一、Drupal 11：队列 API

最近我经常谈论 Drupal 中的批量处理 API，也提到它是基于队列 API 构建的，但还没深入探讨过。我之前写过关于 Drupal 7 中队列 API 的文章，现在想更新一下我的理解。

队列是一种遵循“先进先出”（FIFO）原则的数据结构，也就是按照添加到队列的顺序处理元素。这种系统用途广泛，在异步数据处理方面尤为关键。在 Drupal 开发中，Drupal 及其众多模块都借助队列系统在后台处理信息。

队列和批量处理有所不同。批量处理适用于对时间敏感的操作，用户通常期望这些操作能立即执行；而队列更适合需要在后台处理或者无需用户触发的数据分析。

批量处理通常是无状态的，这意味着如果批量处理在执行过程中中断，有时很难从中断点重新开始。不过，要是以正确方式创建批量处理，还是有重新开始的可能，但这种情况并不常见。相比之下，队列在处理这种情况时表现更佳。它会存储队列中的所有元素，并且在处理每个元素时提供多种操作选项。也就是说，如果某个元素处理失败，可以将其重新放回队列，以便稍后处理。

在本文中，我会介绍 Drupal 11 中的队列 API，包括它的使用方法以及使用该 API 时的最佳实践。

二、创建队列

要在 Drupal 里创建队列，得创建一个“queue”服务的实例。这其实是一个工厂，可用于在 Drupal 中创建和管理队列。默认情况下，Drupal 中的所有队列都是数据库队列（通过 queue.database 默认队列工厂处理），不过可以通过配置设置进行更改。

我们会在后续文章中介绍不同类型的队列，现在就假设我们的队列存储在名为“queue”的数据库表中并由其管理。

创建工厂后，可以使用 get() 方法创建一个简单的队列，并传入你希望队列使用的名称。以下是在 Drupal 中获取队列的方法。

/** @var \Drupal\Core\Queue\QueueFactoryInterface $queueFactory */
$queueFactory = \Drupal::service('queue');

/** @var \Drupal\Core\Queue\QueueInterface $queue */
$queue = $this->queueFactory->get('queue_simple_example');

可以用以下代码简化上述操作。

/** @var \Drupal\Core\Queue\QueueInterface $queue */
$queue = \Drupal::queue('queue_simple_example');

创建好队列后，就能使用 createItem() 方法向其中添加元素。传递给 createItem() 方法的信息会以序列化的 PHP 数据形式存储在队列表中，所以只要后续能够重建这些数据，就可以向队列中传递任何信息。

以下是一个使用 PHP stdClass 对象存储数据并创建队列元素的示例。

$item = new \stdClass;
$item->id = 123;
$queue->createItem($item);

现在查看“queue”表，会发现队列元素已经添加成功。

> select * from queue;
+---------+--------------------------------+------------------------------------+--------+------------+
| item_id | name                           | data                               | expire | created    |
+---------+--------------------------------+------------------------------------+--------+------------+
|    5156 | queue_simple_example           | O:8:"stdClass":1:{s:2:"id";i:123;} |      0 | 1735664606 |
+---------+--------------------------------+------------------------------------+--------+------------+

使用该系统的一种更高级方法是创建一个队列数据传输对象（DTO）类，用于在队列中存储数据。这比使用 stdClass 对象设置数据更好，因为能更好地控制对象存储信息的方式以及在序列化和反序列化过程中的行为。我会在后续文章中详细介绍这种技术，目前我们已经在队列中创建了一个元素。

如果想检查队列中当前有多少个元素，可以使用 numberOfItems() 方法获取该信息。

$number = $queue->numberOfItems()

调用一次 createItem() 方法后，队列中应该只有一个元素。

接下来，我们将对这个队列元素进行处理。

三、队列工作器

向队列中添加元素后，需要一种方法来处理这些元素。队列工作器是一种 Drupal 插件，用于处理队列元素中的数据。在 Drupal 模块开发中，经常会使用到这种插件来实现特定功能。

要创建队列工作器，需要在自定义模块的 src/Plugin/QueueWorker 目录中创建一个继承自 Drupal\Core\Queue\QueueWorkerBase 的类。该类的定义需要包含一个注解，用于告知 Drupal 这是一个队列工作器以及它作用于哪个队列。

延续上面的“queue_simple_example”队列，我们可以这样创建注解。

/**
 * 用于 queue_simple_example 的队列工作器。
 *
 * @QueueWorker(
 *   id = "queue_simple_example",
 *   title = @Translation("简单队列示例的队列工作器。"),
 *   cron = {"time" = 60}
 * )
 */

“title”是队列的可读标题，通常不会显示给用户。“cron”部分指定了在 cron 任务运行期间处理队列的时间，这里设置为 60 秒。

该类必须包含一个名为 processItem() 的方法，该方法接受一个 $data 参数。这个参数是创建队列元素时添加到队列中的数据，需要在该方法中对其进行处理。队列处理器在处理队列时，会调用队列的 claimItem() 方法，该方法会返回一个要处理的元素，并为该元素设置一个租赁时间，确保该元素从数据库中取出后不会被重复处理。

以下是一个非常简单的队列工作器类的实现，用于处理“queue_simple_example”队列中的元素。我们还实现了 ContainerFactoryPluginInterface 接口，以便将不同的服务注入到插件中，从而执行所需的工作。

<?php

namespace Drupal\queue_simple_example\Plugin\QueueWorker;

use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\Core\StringTranslation\StringTranslationTrait;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * 用于 queue_simple_example 的队列工作器。
 *
 * @QueueWorker(
 *   id = "queue_simple_example",
 *   title = @Translation("简单队列示例的队列工作器。"),
 *   cron = {"time" = 60}
 * )
 */
class QueueExampleWorker extends QueueWorkerBase implements ContainerFactoryPluginInterface {

  use StringTranslationTrait;

  /**
   * 日志记录器工厂。
   *
   * @var \Drupal\Core\Logger\LoggerChannelInterface
   */
  protected $logger;

  /**
   * {@inheritDoc}
   */
  public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition) {
    $instance = new self($configuration, $plugin_id, $plugin_definition);
    $instance->logger = $container->get('logger.channel.queue_simple_example');
    return $instance;
  }

  /**
   * {@inheritDoc}
   */
  public function processItem($data) {
    // 在这里处理队列元素，然后创建一条日志消息。
    $this->logger->info($this->t('已处理简单队列元素 @id', ['@id' => $data->id]));
  }

}

processItem() 方法中的具体操作取决于具体需求，可能需要注入相关的服务来正确处理队列元素。

上述 processItem() 方法只是简单地处理成功并返回，没有引发任何错误。这意味着上游的队列处理代码会认为一切正常，并将元素从队列中移除。如果处理过程中出现问题，可以抛出异常，让队列处理器将元素重新放回队列中，以便稍后处理。

在 processItem() 方法中抛出不同类型的异常会对队列元素产生不同的影响。

• \Drupal\Core\Queue\DelayedRequeueException - 如果队列实现了 \Drupal\Core\Queue\DelayableQueueInterface 接口，则使用 delayItem() 方法将队列元素的执行延迟一段时间（作为异常的一部分传递）。如果队列不支持 DelayableQueueInterface 接口，则元素会在队列中暂时保留，默认保留一小时。
• \Drupal\Core\Queue\RequeueException - 元素会被重新添加到队列中，租赁时间会重置，这意味着队列管理器下次处理到该元素时会立即处理。
• \Drupal\Core\Queue\SuspendQueueException - 元素会被重新添加到队列中，并且队列执行会停止。
• \Exception - 元素会被重新添加到队列中以便稍后处理，并记录一条错误日志。

队列在处理其中的元素时，可以抛出上述异常来对队列元素进行不同的操作。所有这些异常都会将元素重新放回队列中，因此如果不想这样做，就不要抛出异常。这样队列处理器会将元素从队列中删除，并继续处理其他元素。

我已经介绍了队列的处理，那么具体该如何操作呢？

四、处理队列

所有队列都会在 cron 任务运行时由 Drupal 自动处理。

这意味着可以通过以下方式处理队列：

• 通过“/admin/config/system/cron”页面的 cron 处理器表单。
• 通过状态页面上的“运行 cron”链接。
• 使用 Drush 命令“drush cron”。

这些方法都会触发 Drupal 从队列中获取当前待处理的元素，并使用相应的队列工作器进行处理。队列工作器注解中设置的时间参数决定了队列处理的时长，一旦达到该限制，Drupal 会自动停止队列处理。

还可以使用 Queue UI 模块来检查和处理队列元素。这个模块可以方便地查看系统中所有队列的元素数量，还可以使用批量处理方式处理队列元素。

五、队列的用途

在 Drupal 中使用队列有很多原因。

Drupal 的核心系统在检查更新时会使用队列 API，因为这种检查不一定需要用户手动触发，可以将其添加到队列中，按照每天或每周的计划进行处理（可在更新管理器设置中配置）。Drupal 升级时，队列系统也能在其中发挥作用，确保相关任务有序进行。

还能看到许多第三方模块也在使用队列系统来执行任务。例如，Warmer 模块会使用队列系统为网站上的不同对象创建缓存，将需要“预热”的元素添加到队列中，然后在 cron 任务运行时进行处理。

以下是一些在队列中处理元素的思路：

• 将任何可能耗时较长的操作添加到队列中，稍后再进行处理。例如，如果要删除网站上的所有分类词汇，可以将所有词汇 ID 添加到队列中，然后在队列处理过程中进行删除操作。根据网站上词汇的数量，这个过程可能需要几秒到几天的时间，但使用队列可以确保每个元素都会按顺序处理。
• 使用队列进行与 API 之间的元素同步非常有用，特别是在不需要用户输入的情况下。例如，假设用户更新个人资料时需要向某个 API 发送详细信息，比如将时事通讯设置推送到第三方服务。当用户保存偏好设置时，可以将一个元素添加到队列中，以便在下一次 cron 任务运行时获取并同步他们的时事通讯设置。这样用户就不需要等待偏好设置立即保存，同时也提高了系统的可靠性。如果 API 出现故障，队列中的元素在处理过程中会抛出异常，并重新放回队列中。当 API 恢复正常后，队列会正确处理更新，用户不会察觉到任何问题。

自从 Drupal 7 引入队列系统以来，我就一直在使用它。实际上，我公司曾为一个客户实现过一个比较复杂的队列系统，该客户需要与一个（非常）缓慢的 API 系统进行交互。这个系统是一个计费平台，会从 API 获取详细信息，并允许用户在线支付账单。问题在于，有些用户在系统中有很多账户，对于一些大型账户，实时获取这些信息可能需要长达 25 分钟。显然，让用户等待这么长时间来加载账户页面是不可接受的，因此我们需要想一个解决方案。

解决这个问题的方法是识别系统中拥有最多账户的用户，并使用一系列队列在夜间预先缓存他们的账户详细信息。这意味着我们可以从 API 为这些用户获取信息，并将其存储在 Drupal 缓存系统中，这样当用户加载他们的账户时，数据会从缓存中获取，而不是直接从 API 获取。当用户更改账户信息时，我们会使更改的项目的缓存失效，并立即重新同步，而无需再次获取所有数据。我公司与客户合作，微调了触发预缓存步骤的账户数量阈值，这个方案解决了一个潜在的大问题，使平台能够快速加载。

基本的经验法则是，任何可以异步处理的任务列表都可以由队列工作器处理。

六、结论

本文介绍了在 Drupal 中使用队列 API 的基础知识。还有一些细节我在这里没有详细介绍，我公司会在后续关于这个主题的文章中进行探讨。

批量处理和队列在 Drupal 中都有各自的应用场景。队列系统适用于不需要立即处理，但仍需及时处理的数据。批量处理系统则用于需要立即处理数据且用户正在等待处理结果的情况。如果不小心，在需要处理大量数据时，批量处理系统可能会出现性能问题。当需要确保所有数据都能被处理，且不需要立即得到结果时，使用队列来存储和处理数据通常是更好的选择。

本文中展示的所有代码都可以在一个 GitHub 仓库中找到，该仓库包含了一系列队列系统的实际应用示例。这个示例模块的工作方式类似于批量处理 API 示例模块，可以选择查看不同的示例。本文中的代码位于名为 queue_simple_example 的子模块中。该模块中的任何操作都不会在未事先告知的情况下删除或更改数据，因此可以放心下载并尝试使用。