一、Drupal 11：使用批量处理 API 进行批量处理介绍

注意：这篇文章发布已超过两年，因此其中包含的信息可能已过时。如果您发现了问题，请留下评论，我公司会尽力更正。

2024 年 8 月 18 日 - 阅读时长 38 分钟

批量处理 API 是 Drupal 中一项强大的功能，它能将复杂或耗时的任务拆分成更小的部分。

例如，假设您要运行一个函数，该函数会遍历 Drupal 网站上的每一个页面并执行特定操作，比如从页面中移除特定作者、删除文本中的链接或者删除某些分类术语等。您可能会创建一个小循环，逐个加载所有页面并对这些页面执行操作。

对于页面数量较少（少于 100 个）的网站，这样做通常没问题。但如果网站有 10,000 个页面甚至 100 万个页面，情况就不同了。您的小循环很快就会达到 PHP 执行时间或内存限制，导致脚本终止。而且您很难知道循环在处理数据时进行到了哪一步，如果尝试重新启动循环，情况也会变得复杂。

Drupal 中的批量处理 API 可以解决这些问题。它不会一次性运行单个进程来更改所有页面，而是将任务拆分成多个部分。当批量处理运行时，一系列较小的任务（例如每次处理 50 个页面）会逐步执行，直到任务完成。这意味着您不会达到 PHP 的内存或超时限制，任务能够成功且可预测地完成。批量处理 API 允许操作通过许多小的页面请求来运行，而不是在单个页面请求中执行，每个小请求都会逐步完成一部分任务，直到整个任务结束。

这种技术可应用于各种不同的场景，在 Drupal 开发以及 Drupal 模块开发中，许多贡献模块都利用了这一功能，以防止进程耗时过长。

我喜欢用一个比喻来解释批量处理过程，就像食物挑战。在我的家乡康格尔顿，有一家名为“贝尔·格里尔斯”的咖啡馆，它举办了一项“贝尔·格里尔斯的灰熊早餐三明治挑战”。这个三明治重达 2.7 公斤，里面有 6 根香肠、6 片培根、4 个鸡蛋、4 片土豆华夫饼、豆子，上面还盖着奶酪。

一次性吃完这个早餐三明治挑战肯定很困难，但如果在几天内分 100 小餐慢慢吃完，就容易多了。这就是批量处理的原理，它将大量的项目拆分成更小的块，这样就更易于处理。

本文是一系列关于批量处理 API 各个方面以及如何使用它的文章中的第一篇。在本文中，我们将介绍核心的批量处理 API 以及如何进行第一次批量运行的设置。

二、批量处理过程

Drupal 中的批量处理过程涉及以下三个步骤。

• 启动步骤 - 这是批量处理开始的地方。最好从某种操作（如控制器、表单或 Drush 命令）来启动批量处理，这样能确保批量处理不受阻碍地进行。当批量处理启动时，网站会重定向到路径 /batch，所以您需要确保这是操作或提交处理程序中最后执行的操作。
• 处理步骤 - 批量处理初始化后，就会开始运行批量处理本身。处理步骤的数量可以在启动步骤中设置，您也可以设置一个步骤，让该步骤重复运行，直到任务完成。在处理步骤中，您可以跟踪进度，包括已处理的项目数量或发生的错误数量，也可以运行多个不同的步骤来执行不同的操作。
• 完成步骤 - 最后一步是完成步骤。在这一步中，您可以记录批量处理中发生的情况，并可选地重定向到网站上的另一个页面。

使用批量处理 API 的复杂性主要在于如何设置处理步骤。初始化批量运行有几种不同的方式，您创建的处理过程将取决于您要完成的任务。

批量处理 API 的核心是 BatchBuilder 类，下面让我们先来了解一下这个类。

三、BatchBuilder 类

Drupal 8 及更高版本中批量处理 API 的核心是 BatchBuilder 类。使用这个类，我们可以创建需要传递给 batch_set() 方法的参数，而 batch_set() 方法正是开始批量操作的地方。

像这样创建一个 BatchBuilder 对象。

use Drupal\Core\Batch\BatchBuilder;
$batch = new BatchBuilder();

BatchBuilder 对象有许多不同的方法，您可以使用这些方法来创建所需的批量设置和操作。以下是可用方法的列表。

• setTitle() - 设置批量处理页面的标题。
• setFinishedCallback() - 设置批量操作完成后要运行的可调用代码，用于记录成功（或错误）并将用户重定向。
• setInitMessage() - 设置处理初始化时显示的消息。
• setProgressMessage() - 设置批量运行期间显示的进度消息（如果没有设置其他消息）。
• setErrorMessage() - 设置批量处理过程中发生错误时要显示的错误消息。
• setFile() - 这允许您设置包含回调函数（即批量操作和完成操作的函数）的文件的位置。此路径应相对于网站的 base_path()，因此应使用 \Drupal\Core\Extension\ExtensionList::getPath() 方法来构建。默认情况下，该路径为 “[模块名称].module”，但如果您明确指定了批量回调函数的位置，则此设置将不会被使用，因为 PHP 已经知道回调函数的位置。
• setLibraries() - 设置批量处理时要使用的库，这些库将包含在批量处理页面中，默认情况下会包含 core/drupal.batch 库。
• setUrlOptions() - 设置要添加到重定向 URL 的选项。
• setProgressive() - 此设置将批量处理改为逐步运行。正常的批量处理应该以逐步的方式运行，这意味着使用多个请求来处理批量操作。您可以关闭此设置，强制批量处理在单个操作中运行。虽然这似乎违背了运行批量处理的初衷，但有时它可能会很有用。例如，如果您知道批量运行的任务非常小，您可以启用此设置，以防止 Drupal 多次启动。
• setQueue() - 这是一个高级设置，可用于更改批量处理系统运行时使用的底层队列存储系统。对于正常的批量处理，此设置通常设置为 \Drupal\Core\Queue\Batch，但如果 progressive 设置为 true，则可以将其设置为 \Drupal\Core\Queue\BatchMemory。重要的是要记住，批量处理系统使用 Drupal 队列 API 工作，我们设置的每个操作都是队列中的一个项目。
• addOperation() - 使用此方法设置批量处理过程中要运行的批量操作的回调函数。
• toArray() - 这是一个实用方法，将对象的所有设置转换为数组，用于将批量信息传递给 Drupal 批量运行器。

例如，要设置一个最小的批量处理过程，您可以像这样设置批量操作对象。

$batch = new BatchBuilder();
$batch->setTitle('Running batch process.')
  ->setFinishCallback([self::class, 'batchFinished'])
  ->setInitMessage('Commencing')
  ->setProgressMessage('Processing...')
  ->setErrorMessage('An error occurred during processing.');

然后，您只需要设置要运行的操作。作为一个例子，这里有一个批量操作，它以每组 100 个项目的方式对 1 到 1000 的数字进行计数。

// 创建 10 组，每组 100 个项目。
$chunks = array_chunk(range(1, 1000), 100);

// 将数组中的每个块处理为批量处理中的操作。
foreach ($chunks as $id => $chunk) {
  $args = [
    $id,
    $chunk,
  ];
  $batch->addOperation([self::class, 'batchProcess'], $args);
}

批量处理设置完成后，我们使用 batch_set() 方法并传入 toArray() 方法的输出来启动批量运行。

batch_set($batch->toArray());

这将启动批量处理，并运行我们在 addOperation() 方法调用中设置的操作，然后在 setFinishCallback() 方法中设置的方法处结束。

四、批量处理方法

批量处理方法是进行实际处理的地方，也是批量运行的主体。该方法的名称和参数取决于您在设置批量处理时调用 addOperation() 方法时使用的参数数组。

在批量处理设置代码中，我们多次调用了 batchProcess() 方法，并传入了一个长度为 2 的参数数组。下面是单独拿出的调用代码。

$args = [
  $id,
  $chunk,
];
$batch->addOperation([self::class, 'batchProcess'], $args);

这意味着处理方法与当前类在同一个类中，并且具有以下形式。我们将逐步构建该方法，使其包含我们所需的所有部分。

public static function batchProcess(int $batchId, array $chunk, array &$context): void {
}

第一个参数是 $id 的值，第二个参数是 $chunk 变量，并且我们总是会得到一个名为 $context 的最后一个参数，该参数通过引用传递。$context 变量是批量运行所有内部跟踪信息的存储位置，我们可以使用它来初始化变量、报告进度，甚至在完成后停止批量处理。

当我们首次启动批量处理时，$context 数组将如下所示。

Array(
    [sandbox] => Array()
    [results] => Array()
    [finished] => 1
    [message] =>
)

这个数组的各个组件具有以下功能。

• sandbox - 这仅在批量处理方法中使用，通常用于跟踪批量运行的进度，或确定批量中的最大元素数量。批量处理完成后，这个数组将被丢弃。
• results - 批量处理方法使用这个数组来跟踪批量运行的进度。不同之处在于，这个数组会传递给完成回调方法，这使我们能够报告批量处理的情况。因此，这个数组通常用于存储成功或失败操作的数量。您添加到这个数组部分的内容取决于您希望在完成输出中显示的内容。
• finished - 这是一个特殊值，批量处理系统使用它来判断批量处理是否完成。如果您将其设置为小于 1 的值，Drupal 将再次调用批量处理方法以完成批量处理。这个值非常强大，但仅在开放式批量处理过程中起作用。如果您使用特定数量的项目和固定数量的操作来设置批量处理过程，则不会使用这个标志。我将在后续文章中详细介绍这个设置。
• message - 为了向用户传达进度，您可以向这个数组变量设置一条消息，这条消息将显示在批量处理页面上（连同进度条）。

当我们首次启动批量处理时，sandbox 和 results 数组项中没有任何信息，因此我们首先在处理方法中设置这些值。由于我们也知道当前正在运行的批量处理的一些信息，我们还可以向 $context 数组的 message 参数添加内容。

public static function batchProcess(int $batchId, array $chunk, array &$context): void {
  if (!isset($context['sandbox']['progress'])) {
    $context['sandbox']['progress'] = 0;
    $context['sandbox']['max'] = 1000;
  }
  if (!isset($context['results']['updated'])) {
    $context['results']['updated'] = 0;
    $context['results']['skipped'] = 0;
    $context['results']['failed'] = 0;
    $context['results']['progress'] = 0;
  	$context['results']['process'] = 'Chunk batch completed';
  }

  // 进度条上方的消息。
  $context['message'] = t('Processing batch #@batch_id batch size @batch_size for total @count items.', [
    '@batch_id' => number_format($batchId),
    '@batch_size' => number_format(count($chunk)),
    '@count' => number_format($context['sandbox']['max']),
  ]);

  // 处理块。 
}

接下来要添加的是处理数组项块。

为了避免批量操作对网站造成任何破坏性影响，我决定只是遍历每个块中的项目，并让进程休眠几毫秒，以模拟网站上发生的操作。这意味着您可以随意多次运行这个批量调用，而不会导致网站添加（或删除）大量内容。在本系列文章的后续部分，我将介绍更具体的机制，展示如何创建节点。

public static function batchProcess(int $batchId, array $chunk, array &$context): void {
  if (!isset($context['sandbox']['progress'])) {
    $context['sandbox']['progress'] = 0;
    $context['sandbox']['max'] = 1000;
  }
  if (!isset($context['results']['updated'])) {
    $context['results']['updated'] = 0;
    $context['results']['skipped'] = 0;
    $context['results']['failed'] = 0;
    $context['results']['progress'] = 0;
    $context['results']['process'] = 'Form batch completed';
  }

  // 跟踪进度。
  $context['results']['progress'] += count($chunk);

  // 进度条上方的消息。
  $context['message'] = t('Processing batch #@batch_id batch size @batch_size for total @count items.', [
    '@batch_id' => number_format($batchId),
    '@batch_size' => number_format(count($chunk)),
    '@count' => number_format($context['sandbox']['max']),
  ]);

  foreach ($chunk as $number) {
    // 休眠一段时间（利用 number 变量）以模拟正在进行的工作。
    // 这样做是为了让批量处理需要一定的时间才能完成。
    usleep(4000 + $number);
    // 确定批量处理的结果。我们在这里使用随机参数来模拟批量处理过程中发生的不同情况。
    $result = rand(1, 4);
    switch ($result) {
      case '1':
      case '2':
        $context['results']['updated']++;
        break;

      case '3':
        $context['results']['skipped']++;
        break;

      case '4':
        $context['results']['failed']++;
        break;
    }
  }
}

作为模拟处理批量操作的一部分，我还添加了一个随机函数，该函数会选择一个 1 到 4 之间的数字，并增加 context 数组中 results 部分的项目数。由于这些项目将传递给完成方法，我们可以模拟批量运行中出现的一些问题，并查看其结果。

这基本上就是批量处理方法的全部内容。批量处理 API 将调用我们在开始时设置的每个操作方法，并传入我们为每个操作设置的数组项。完成后，结果将传递给批量完成方法。

五、批量完成方法

批量完成方法是批量操作完成后调用的最后一个函数。该方法接受以下参数。

• $success - 如果所有批量处理 API 任务都成功完成，则为 TRUE。
• $results - 批量处理操作的结果数组。
• $operations - 未完成的操作列表。
• $elapsed - Batch.inc 会贴心地提供处理所花费的时间（以秒为单位）。

利用这些信息，我们可以设置一个非常简单的完成方法。我们只需要查看 $success 变量是否为 true。如果是，则可以通过 Messenger 服务向用户报告批量处理已完成，并记录该事实。如果批量处理失败（无论出于何种原因），我们将其作为错误输出，并传入导致问题的操作。

以下是一个典型的完成方法，基于我们在上述步骤中运行的批量操作。

public static function batchFinished(bool $success, array $results, array $operations, string $elapsed): void {
  // 获取 Messenger 服务，如果批量处理成功或失败，都需要这个服务。
  $messenger = \Drupal::messenger();
  if ($success) {
    // success 变量为 true，这表明批量处理成功（即未发生错误）。
    // 向用户显示成功消息。
    $messenger->addMessage(t('@process processed @count, skipped @skipped, updated @updated, failed @failed in @elapsed.', [
      '@process' => $results['process'],
      '@count' => $results['progress'],
      '@skipped' => $results['skipped'],
      '@updated' => $results['updated'],
      '@failed' => $results['failed'],
      '@elapsed' => $elapsed,
    ]));
    // 记录批量处理成功。
    \Drupal::logger('batch_form_example')->info(
      '@process processed @count, skipped @skipped, updated @updated, failed @failed in @elapsed.', [
        '@process' => $results['process'],
        '@count' => $results['progress'],
        '@skipped' => $results['skipped'],
        '@updated' => $results['updated'],
        '@failed' => $results['failed'],
        '@elapsed' => $elapsed,
      ]);
  }
  else {
    // 发生了错误。$operations 包含未处理的操作。选择最后一个操作并报告发生的情况。
    $error_operation = reset($operations);
    if ($error_operation) {
      $message = t('An error occurred while processing %error_operation with arguments: @arguments', [
        '%error_operation' => print_r($error_operation[0]),
        '@arguments' => print_r($error_operation[1], TRUE),
      ]);
      $messenger->addError($message);
    }
  }
}

请记住，这里的结果数组包含您在批量操作步骤中放入其中的信息。这意味着如果您想执行不同的操作或报告不同的活动，则需要更改此代码以报告不同结果数组的内容。

完成方法中的最后一个要点是返回值，它取决于您从何处启动批量处理。如果您从表单启动批量操作，则表单重定向将被考虑在内，并用于将用户发送到表单中设置的任何位置。如果批量操作从控制器启动，则返回值必须是重定向响应，因为控制器必须返回渲染数组或响应对象。

本质上，如果您从完成方法返回一个重定向响应，那么将使用该响应并将用户重定向，但返回重定向响应是可选的。

六、从表单运行批量处理

从表单启动批量操作是比较常见的做法。这样做意味着我们可以从用户那里接受关于批量处理要执行的操作的参数，同时也能更明确地提醒用户，执行此操作可能会导致一个（潜在）耗时较长的过程。

在表单中设置批量操作其实很简单，在表单类的 submitForm() 处理程序中，我们只需创建一个新的 BatchBuilder 对象并设置批量处理。

public function submitForm(array &$form, FormStateInterface $form_state): void {
  // 创建并设置批量处理构建器对象。
  $batch = new BatchBuilder();
  $batch->setTitle('Running batch process.')
    ->setFinishCallback([self::class, 'batchFinished'])
    ->setInitMessage('Commencing')
    ->setProgressMessage('Processing...')
    ->setErrorMessage('An error occurred during processing.');

  // 创建 10 组，每组 100 个项目。
  $chunks = array_chunk(range(1, 1000), 100);

  // 将数组中的每个块处理为批量处理中的操作。
  foreach ($chunks as $id => $chunk) {
    $args = [
      $id,
      $chunk,
    ];
    $batch->addOperation([self::class, 'batchProcess'], $args);
  }
  batch_set($batch->toArray());

  // 设置表单提交后的重定向，回到表单本身。
  $form_state->setRedirectUrl(new Url($this->getFormId()));
}

由于这是一个表单操作，我们可以使用 $form_state 对象来更改批量处理完成后的重定向。批量处理 API 会识别这一点，并在完成方法调用后将其作为最终目标（假设完成方法本身不返回重定向响应）。

当我们提交此表单时，会看到以下批量处理过程正在运行。

完成后，我们将被重定向回提交的表单，在那里会显示一条消息，告诉我们处理了多少项。

七、何时使用批量处理 API

有许多情况可能需要使用批量处理 API，我在引言中已经暗示了一些，下面是一些例子的列表。

• 对大量不同的内容项执行操作。例如，更新网站上的每个页面或删除大量分类术语。
• 如果您与需要大量操作才能完成任务的 API 进行交互，那么批量处理 API 可能会很有用。这允许您在执行操作时向用户显示进度条，并且通常可以掩盖缓慢的 API 系统，或防止 API 使用户页面超时。
• 如果您想从用户那里接受一个文件并处理结果，使用批量处理 API 通常可以帮助将该文件分解成更小的部分。我曾使用批量处理 API 成功解析了一个包含 100,000 条记录的 CSV 文件。

八、何时不使用批量处理 API

当然，批量处理 API 并非在所有情况下都是最佳选择。如果您想快速处理一批项目并在处理过程中向用户提供反馈，那么批量处理 API 通常是最佳方法。

如果您不需要向用户提供反馈，或者时间尺度不是很重要，那么仅使用队列处理器可能是更好的解决方案。Drupal 中的批量处理 API 是基于队列 API 构建的，所以如果您构建了批量操作，之后转换为队列处理器也并不困难。

九、结论

Drupal 中的批量处理 API 是处理数据的一个非常强大的组件，同时也能为用户提供良好的体验。它避免了长时间的页面加载，并引入了一个不错的进度条，让您知道用户需要等待多久才能完成任务。在 Drupal 11 的开发、Drupal 升级以及 Drupal 模块开发中，Drupal 在一些地方使用了批量处理 API，甚至允许 Drupal 的某些部分（例如更新钩子）通过很少的额外代码与批量处理 API 集成。

这里有很多关于如何设置和使用批量处理 API 的信息，但我这里展示的是使用批量处理的最简单版本。使用上述代码，您可以创建一个表单来运行一个需要几秒钟的批量操作，这应该可以让您试验该 API 并了解它的作用。

在下一篇文章中，我们将介绍如何设置批量运行，使其可以通过表单或 Drush 命令运行。

如果您想查看上述示例的源代码，我公司已经将它们作为一个项目发布，该项目有多个不同的子模块，展示了如何在不同的情况和组合中使用批量处理 API。欢迎查看该项目，并在您自己的项目中使用源代码。另外，如果您能想到对该模块的任何改进建议，请告诉我。

最后，我要感谢相关书籍，我在这里使用的一些代码示例就来自其中。关于 Drupal 批量和队列操作的章节真的很值得一读。